d9b0c6fbf7
Drop pycodestyle for code style checking in favor of black. Black is much faster, stable PEP8 compliant code style checker offering also automatic formatting. It aims to be very stable and produce smallest diffs. It's used by many small and big projects. Running checkstyle with black takes a few seconds with a terse output. Thus, test-checkstyle-diff is no longer necessary. Expand scope of checkstyle to all python files in the repo, replacing test-checkstyle with checkstyle-python. Also, fixstyle-python is now available for automatic style formatting. Note: python virtualenv has been consolidated in test/Makefile, test/requirements*.txt which will eventually be moved to a central location. This is required to simply the automated generation of docker executor images in the CI. Type: improvement Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8 Signed-off-by: Klement Sekera <klement.sekera@gmail.com> Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
543 lines
16 KiB
Python
543 lines
16 KiB
Python
""" test framework utilities """
|
|
|
|
import ipaddress
|
|
import logging
|
|
import socket
|
|
from socket import AF_INET6
|
|
import os.path
|
|
from copy import deepcopy
|
|
from collections import UserDict
|
|
|
|
import scapy.compat
|
|
from scapy.layers.l2 import Ether
|
|
from scapy.layers.inet import IP
|
|
from scapy.layers.inet6 import (
|
|
IPv6,
|
|
IPv6ExtHdrFragment,
|
|
IPv6ExtHdrRouting,
|
|
IPv6ExtHdrHopByHop,
|
|
)
|
|
from scapy.packet import Raw
|
|
from scapy.utils import hexdump
|
|
from scapy.utils6 import in6_mactoifaceid
|
|
|
|
from io import BytesIO
|
|
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 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.
|
|
|
|
:param headline: printed as first line of output
|
|
:param capture: packets to print
|
|
:param limit: limit the print to # of packets
|
|
"""
|
|
if not capture:
|
|
return headline
|
|
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)]
|
|
)
|
|
return "%s\n%s%s" % (headline, body, tail)
|
|
|
|
|
|
def ip4_range(ip4, s, e):
|
|
tmp = ip4.rsplit(".", 1)[0]
|
|
return ("%s.%d" % (tmp, i) for i in range(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(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))
|
|
|
|
|
|
def get_core_path(tempdir):
|
|
return "%s/%s" % (tempdir, get_core_pattern())
|
|
|
|
|
|
def is_core_present(tempdir):
|
|
return os.path.isfile(get_core_path(tempdir))
|
|
|
|
|
|
def get_core_pattern():
|
|
with open("/proc/sys/kernel/core_pattern", "r") as f:
|
|
corefmt = f.read().strip()
|
|
return corefmt
|
|
|
|
|
|
def check_core_path(logger, core_path):
|
|
corefmt = get_core_pattern()
|
|
if corefmt.startswith("|"):
|
|
logger.error(
|
|
"WARNING: redirecting the core dump through a"
|
|
" 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)
|
|
|
|
|
|
class NumericConstant:
|
|
|
|
desc_dict = {}
|
|
|
|
def __init__(self, value):
|
|
self._value = value
|
|
|
|
def __int__(self):
|
|
return self._value
|
|
|
|
def __long__(self):
|
|
return self._value
|
|
|
|
def __str__(self):
|
|
if self._value in self.desc_dict:
|
|
return self.desc_dict[self._value]
|
|
return ""
|
|
|
|
|
|
class Host:
|
|
"""Generic test host "connected" to VPPs interface."""
|
|
|
|
@property
|
|
def mac(self):
|
|
"""MAC address"""
|
|
return self._mac
|
|
|
|
@property
|
|
def bin_mac(self):
|
|
"""MAC address"""
|
|
return mac_pton(self._mac)
|
|
|
|
@property
|
|
def ip4(self):
|
|
"""IPv4 address - string"""
|
|
return self._ip4
|
|
|
|
@property
|
|
def ip4n(self):
|
|
"""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"""
|
|
return self._ip6
|
|
|
|
@property
|
|
def ip6n(self):
|
|
"""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"""
|
|
return self._ip6_ll
|
|
|
|
@property
|
|
def ip6n_ll(self):
|
|
"""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
|
|
)
|
|
else:
|
|
return False
|
|
|
|
def __ne__(self, other):
|
|
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,
|
|
)
|
|
|
|
def __hash__(self):
|
|
return hash(self.__repr__())
|
|
|
|
def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
|
|
self._mac = mac
|
|
self._ip4 = ip4
|
|
self._ip6 = ip6
|
|
self._ip6_ll = ip6_ll
|
|
|
|
|
|
class L4_Conn:
|
|
"""L4 'connection' tied to two VPP interfaces"""
|
|
|
|
def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
|
|
self.testcase = testcase
|
|
self.ifs = [None, None]
|
|
self.ifs[0] = if1
|
|
self.ifs[1] = if2
|
|
self.address_family = af
|
|
self.l4proto = l4proto
|
|
self.ports = [None, None]
|
|
self.ports[0] = port1
|
|
self.ports[1] = port2
|
|
self
|
|
|
|
def pkt(self, side, l4args={}, payload="x"):
|
|
is_ip6 = 1 if self.address_family == AF_INET6 else 0
|
|
s0 = side
|
|
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]}
|
|
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)
|
|
)
|
|
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))
|
|
self.ifs[1 - side].enable_capture()
|
|
self.testcase.pg_start()
|
|
|
|
def recv(self, side):
|
|
p = self.ifs[side].wait_for_packet(1)
|
|
return p
|
|
|
|
def send_through(self, side, flags=None, payload=""):
|
|
self.send(side, flags, payload)
|
|
p = self.recv(1 - side)
|
|
return p
|
|
|
|
def send_pingpong(self, side, flags1=None, flags2=None):
|
|
p1 = self.send_through(side, flags1)
|
|
p2 = self.send_through(1 - side, flags2)
|
|
return [p1, p2]
|
|
|
|
|
|
class L4_CONN_SIDE:
|
|
L4_CONN_SIDE_ZERO = 0
|
|
L4_CONN_SIDE_ONE = 1
|
|
|
|
|
|
def fragment_rfc791(packet, fragsize, logger=null_logger):
|
|
"""
|
|
Fragment an IPv4 packet per RFC 791
|
|
:param packet: packet to fragment
|
|
:param fragsize: size at which to fragment
|
|
:note: IP options are not supported
|
|
:returns: list of fragments
|
|
"""
|
|
logger.debug(ppp("Fragmenting packet:", packet))
|
|
packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
|
|
if len(packet[IP].options) > 0:
|
|
raise Exception("Not implemented")
|
|
if len(packet) <= fragsize:
|
|
return [packet]
|
|
|
|
pre_ip_len = len(packet) - len(packet[IP])
|
|
ip_header_len = packet[IP].ihl * 4
|
|
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 = int((fragsize - pre_ip_len - ihl * 4) / 8)
|
|
fo = packet[IP].frag
|
|
|
|
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[IP].len = otl - nfb * 8
|
|
p[IP].frag = fo + nfb
|
|
del p[IP].chksum
|
|
|
|
more_fragments = fragment_rfc791(p, fragsize, logger)
|
|
pkts.extend(more_fragments)
|
|
|
|
return pkts
|
|
|
|
|
|
def fragment_rfc8200(packet, identification, fragsize, logger=null_logger):
|
|
"""
|
|
Fragment an IPv6 packet per RFC 8200
|
|
:param packet: packet to fragment
|
|
:param fragsize: size at which to fragment
|
|
:note: IP options are not supported
|
|
:returns: list of fragments
|
|
"""
|
|
packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
|
|
if len(packet) <= fragsize:
|
|
return [packet]
|
|
logger.debug(ppp("Fragmenting packet:", packet))
|
|
pkts = []
|
|
counter = 0
|
|
routing_hdr = None
|
|
hop_by_hop_hdr = None
|
|
upper_layer = None
|
|
seen_ipv6 = False
|
|
ipv6_nr = -1
|
|
l = packet.getlayer(counter)
|
|
while l is not None:
|
|
if l.__class__ is IPv6:
|
|
if seen_ipv6:
|
|
# ignore 2nd IPv6 header and everything below..
|
|
break
|
|
ipv6_nr = counter
|
|
seen_ipv6 = True
|
|
elif l.__class__ is IPv6ExtHdrFragment:
|
|
raise Exception("Already fragmented")
|
|
elif l.__class__ is IPv6ExtHdrRouting:
|
|
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")
|
|
):
|
|
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)
|
|
)
|
|
|
|
if upper_layer is None:
|
|
raise Exception("Upper layer header not found in IPv6 packet")
|
|
|
|
last_per_fragment_hdr = ipv6_nr
|
|
if routing_hdr is None:
|
|
if hop_by_hop_hdr is not None:
|
|
last_per_fragment_hdr = hop_by_hop_hdr
|
|
else:
|
|
last_per_fragment_hdr = routing_hdr
|
|
logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
|
|
|
|
per_fragment_headers = packet.copy()
|
|
per_fragment_headers[last_per_fragment_hdr].remove_payload()
|
|
logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
|
|
|
|
ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
|
|
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))
|
|
|
|
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].plen
|
|
del p[IPv6].nh
|
|
p = p / fragment_ext_hdr
|
|
del p[IPv6ExtHdrFragment].nh
|
|
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__(scapy.compat.raw(p))
|
|
logger.debug(ppp("Fragment %s:" % len(pkts), p))
|
|
pkts.append(p)
|
|
offset = first_payload_len_nfb * 8
|
|
logger.debug("Offset after first fragment: %s" % offset)
|
|
while len(hex_payload) > offset:
|
|
p = per_fragment_headers
|
|
del p[IPv6].plen
|
|
del p[IPv6].nh
|
|
p = p / fragment_ext_hdr
|
|
del p[IPv6ExtHdrFragment].nh
|
|
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 = int(offset / 8)
|
|
p[IPv6ExtHdrFragment].m = 1
|
|
p = p.__class__(scapy.compat.raw(p))
|
|
logger.debug(ppp("Fragment %s:" % len(pkts), p))
|
|
pkts.append(p)
|
|
offset = offset + l_nfb * 8
|
|
|
|
pkts[-1][IPv6ExtHdrFragment].m = 0 # reset more-flags in last fragment
|
|
|
|
return pkts
|
|
|
|
|
|
def reassemble4_core(listoffragments, return_ip):
|
|
buffer = BytesIO()
|
|
first = listoffragments[0]
|
|
buffer.seek(20)
|
|
for pkt in listoffragments:
|
|
buffer.seek(pkt[IP].frag * 8)
|
|
buffer.write(bytes(pkt[IP].payload))
|
|
first.len = len(buffer.getvalue()) + 20
|
|
first.flags = 0
|
|
del first.chksum
|
|
if return_ip:
|
|
header = bytes(first[IP])[:20]
|
|
return first[IP].__class__(header + buffer.getvalue())
|
|
else:
|
|
header = bytes(first[Ether])[:34]
|
|
return first[Ether].__class__(header + buffer.getvalue())
|
|
|
|
|
|
def reassemble4_ether(listoffragments):
|
|
return reassemble4_core(listoffragments, False)
|
|
|
|
|
|
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)
|