test/util.py - fdio/vpp - Gitiles

 """ 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)
	""" 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)