1 """ test framework utilities """
6 from socket import AF_INET6
9 from copy import deepcopy
10 from collections import UserDict
13 from scapy.layers.l2 import Ether
14 from scapy.layers.inet import IP
15 from scapy.layers.inet6 import (
21 from scapy.packet import Raw
22 from scapy.utils import hexdump
23 from scapy.utils6 import in6_mactoifaceid
25 from io import BytesIO
26 from vpp_papi import mac_pton
28 # Set up an empty logger for the testcase that can be overridden as necessary
29 null_logger = logging.getLogger("VppTestCase.util")
30 null_logger.addHandler(logging.NullHandler())
34 return packet.__repr__()
37 def ppp(headline, packet):
38 """Return string containing headline and output of scapy packet.show()"""
39 return "%s\n%s\n\n%s\n" % (
41 hexdump(packet, dump=True),
42 packet.show(dump=True),
46 def ppc(headline, capture, limit=10):
47 """Return string containing ppp() printout for a capture.
49 :param headline: printed as first line of output
50 :param capture: packets to print
51 :param limit: limit the print to # of packets
56 if limit < len(capture):
57 tail = "\nPrint limit reached, %s out of %s packets printed" % (
62 [ppp("Packet #%s:" % count, p) for count, p in zip(range(0, limit), capture)]
64 return "%s\n%s%s" % (headline, body, tail)
67 def ip4_range(ip4, s, e):
68 tmp = ip4.rsplit(".", 1)[0]
69 return ("%s.%d" % (tmp, i) for i in range(s, e))
72 def mcast_ip_to_mac(ip):
73 ip = ipaddress.ip_address(ip)
74 if not ip.is_multicast:
75 raise ValueError("Must be multicast address.")
78 mcast_mac = "01:00:5e:%02x:%02x:%02x" % (
79 (ip_as_int >> 16) & 0x7F,
80 (ip_as_int >> 8) & 0xFF,
84 mcast_mac = "33:33:%02x:%02x:%02x:%02x" % (
85 (ip_as_int >> 24) & 0xFF,
86 (ip_as_int >> 16) & 0xFF,
87 (ip_as_int >> 8) & 0xFF,
93 # wrapper around scapy library function.
95 euid = in6_mactoifaceid(str(mac))
96 addr = "fe80::" + euid
100 def ip6_normalize(ip6):
101 return socket.inet_ntop(socket.AF_INET6, socket.inet_pton(socket.AF_INET6, ip6))
104 def get_core_path(tempdir):
105 return "%s/%s" % (tempdir, get_core_pattern())
108 def is_core_present(tempdir):
109 return os.path.isfile(get_core_path(tempdir))
112 def get_core_pattern():
113 if platform.uname().system == "FreeBSD":
116 corefmt = sysctl.filter("kern.corefile")[0].value
117 elif platform.uname().system == "Linux":
118 with open("/proc/sys/kernel/core_pattern", "r") as f:
119 corefmt = f.read().strip()
123 def check_core_path(logger, core_path):
124 corefmt = get_core_pattern()
125 if corefmt.startswith("|"):
127 "WARNING: redirecting the core dump through a"
128 " filter may result in truncated dumps."
131 " You may want to check the filter settings"
132 " or uninstall it and edit the"
133 " /proc/sys/kernel/core_pattern accordingly."
135 logger.error(" current core pattern is: %s" % corefmt)
138 class NumericConstant:
141 def __init__(self, value):
151 if self._value in self.desc_dict:
152 return self.desc_dict[self._value]
157 """Generic test host "connected" to VPPs interface."""
167 return mac_pton(self._mac)
171 """IPv4 address - string"""
176 """IPv4 address of remote host - raw, suitable as API parameter."""
177 return socket.inet_pton(socket.AF_INET, self._ip4)
181 """IPv6 address - string"""
186 """IPv6 address of remote host - raw, suitable as API parameter."""
187 return socket.inet_pton(socket.AF_INET6, self._ip6)
191 """IPv6 link-local address - string"""
196 """IPv6 link-local address of remote host -
197 raw, suitable as API parameter."""
198 return socket.inet_pton(socket.AF_INET6, self._ip6_ll)
200 def __eq__(self, other):
201 if isinstance(other, Host):
203 self.mac == other.mac
204 and self.ip4 == other.ip4
205 and self.ip6 == other.ip6
206 and self.ip6_ll == other.ip6_ll
211 def __ne__(self, other):
212 return not self.__eq__(other)
215 return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (
223 return hash(self.__repr__())
225 def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
229 self._ip6_ll = ip6_ll
233 """L4 'connection' tied to two VPP interfaces"""
235 def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
236 self.testcase = testcase
237 self.ifs = [None, None]
240 self.address_family = af
241 self.l4proto = l4proto
242 self.ports = [None, None]
243 self.ports[0] = port1
244 self.ports[1] = port2
247 def pkt(self, side, l4args={}, payload="x"):
248 is_ip6 = 1 if self.address_family == AF_INET6 else 0
251 src_if = self.ifs[s0]
252 dst_if = self.ifs[s1]
254 IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
255 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6),
257 merged_l4args = {"sport": self.ports[s0], "dport": self.ports[s1]}
258 merged_l4args.update(l4args)
260 Ether(dst=src_if.local_mac, src=src_if.remote_mac)
262 / self.l4proto(**merged_l4args)
267 def send(self, side, flags=None, payload=""):
269 if flags is not None:
270 l4args["flags"] = flags
271 self.ifs[side].add_stream(self.pkt(side, l4args=l4args, payload=payload))
272 self.ifs[1 - side].enable_capture()
273 self.testcase.pg_start()
275 def recv(self, side):
276 p = self.ifs[side].wait_for_packet(1)
279 def send_through(self, side, flags=None, payload=""):
280 self.send(side, flags, payload)
281 p = self.recv(1 - side)
284 def send_pingpong(self, side, flags1=None, flags2=None):
285 p1 = self.send_through(side, flags1)
286 p2 = self.send_through(1 - side, flags2)
291 L4_CONN_SIDE_ZERO = 0
295 def fragment_rfc791(packet, fragsize, logger=null_logger):
297 Fragment an IPv4 packet per RFC 791
298 :param packet: packet to fragment
299 :param fragsize: size at which to fragment
300 :note: IP options are not supported
301 :returns: list of fragments
303 logger.debug(ppp("Fragmenting packet:", packet))
304 packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
305 if len(packet[IP].options) > 0:
306 raise Exception("Not implemented")
307 if len(packet) <= fragsize:
310 pre_ip_len = len(packet) - len(packet[IP])
311 ip_header_len = packet[IP].ihl * 4
312 hex_packet = scapy.compat.raw(packet)
313 hex_headers = hex_packet[: (pre_ip_len + ip_header_len)]
314 hex_payload = hex_packet[(pre_ip_len + ip_header_len) :]
318 otl = len(packet[IP])
319 nfb = int((fragsize - pre_ip_len - ihl * 4) / 8)
322 p = packet.__class__(hex_headers + hex_payload[: nfb * 8])
325 p[IP].len = ihl * 4 + nfb * 8
329 p = packet.__class__(hex_headers + hex_payload[nfb * 8 :])
330 p[IP].len = otl - nfb * 8
331 p[IP].frag = fo + nfb
334 more_fragments = fragment_rfc791(p, fragsize, logger)
335 pkts.extend(more_fragments)
340 def fragment_rfc8200(packet, identification, fragsize, logger=null_logger):
342 Fragment an IPv6 packet per RFC 8200
343 :param packet: packet to fragment
344 :param fragsize: size at which to fragment
345 :note: IP options are not supported
346 :returns: list of fragments
348 packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
349 if len(packet) <= fragsize:
351 logger.debug(ppp("Fragmenting packet:", packet))
355 hop_by_hop_hdr = None
359 l = packet.getlayer(counter)
361 if l.__class__ is IPv6:
363 # ignore 2nd IPv6 header and everything below..
367 elif l.__class__ is IPv6ExtHdrFragment:
368 raise Exception("Already fragmented")
369 elif l.__class__ is IPv6ExtHdrRouting:
370 routing_hdr = counter
371 elif l.__class__ is IPv6ExtHdrHopByHop:
372 hop_by_hop_hdr = counter
376 and not l.__class__.__name__.startswith("IPv6ExtHdr")
378 upper_layer = counter
379 counter = counter + 1
380 l = packet.getlayer(counter)
383 "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)"
384 % (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer)
387 if upper_layer is None:
388 raise Exception("Upper layer header not found in IPv6 packet")
390 last_per_fragment_hdr = ipv6_nr
391 if routing_hdr is None:
392 if hop_by_hop_hdr is not None:
393 last_per_fragment_hdr = hop_by_hop_hdr
395 last_per_fragment_hdr = routing_hdr
396 logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
398 per_fragment_headers = packet.copy()
399 per_fragment_headers[last_per_fragment_hdr].remove_payload()
400 logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
402 ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
403 hex_payload = scapy.compat.raw(ext_and_upper_layer)
404 logger.debug("Payload length is %s" % len(hex_payload))
405 logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
407 fragment_ext_hdr = IPv6ExtHdrFragment()
408 logger.debug(ppp("Fragment header:", fragment_ext_hdr))
410 len_ext_and_upper_layer_payload = len(ext_and_upper_layer.payload)
411 if not len_ext_and_upper_layer_payload and hasattr(ext_and_upper_layer, "data"):
412 len_ext_and_upper_layer_payload = len(ext_and_upper_layer.data)
415 len(per_fragment_headers)
416 + len(fragment_ext_hdr)
417 + len(ext_and_upper_layer)
418 - len_ext_and_upper_layer_payload
422 "Cannot fragment this packet - MTU too small "
423 "(%s, %s, %s, %s, %s)"
425 len(per_fragment_headers),
426 len(fragment_ext_hdr),
427 len(ext_and_upper_layer),
428 len_ext_and_upper_layer_payload,
433 orig_nh = packet[IPv6].nh
434 p = per_fragment_headers
437 p = p / fragment_ext_hdr
438 del p[IPv6ExtHdrFragment].nh
439 first_payload_len_nfb = int((fragsize - len(p)) / 8)
440 p = p / Raw(hex_payload[: first_payload_len_nfb * 8])
442 p[IPv6ExtHdrFragment].nh = orig_nh
443 p[IPv6ExtHdrFragment].id = identification
444 p[IPv6ExtHdrFragment].offset = 0
445 p[IPv6ExtHdrFragment].m = 1
446 p = p.__class__(scapy.compat.raw(p))
447 logger.debug(ppp("Fragment %s:" % len(pkts), p))
449 offset = first_payload_len_nfb * 8
450 logger.debug("Offset after first fragment: %s" % offset)
451 while len(hex_payload) > offset:
452 p = per_fragment_headers
455 p = p / fragment_ext_hdr
456 del p[IPv6ExtHdrFragment].nh
457 l_nfb = int((fragsize - len(p)) / 8)
458 p = p / Raw(hex_payload[offset : offset + l_nfb * 8])
459 p[IPv6ExtHdrFragment].nh = orig_nh
460 p[IPv6ExtHdrFragment].id = identification
461 p[IPv6ExtHdrFragment].offset = int(offset / 8)
462 p[IPv6ExtHdrFragment].m = 1
463 p = p.__class__(scapy.compat.raw(p))
464 logger.debug(ppp("Fragment %s:" % len(pkts), p))
466 offset = offset + l_nfb * 8
468 pkts[-1][IPv6ExtHdrFragment].m = 0 # reset more-flags in last fragment
473 def reassemble4_core(listoffragments, return_ip):
475 first = listoffragments[0]
477 for pkt in listoffragments:
478 buffer.seek(pkt[IP].frag * 8)
479 buffer.write(bytes(pkt[IP].payload))
480 first.len = len(buffer.getvalue()) + 20
484 header = bytes(first[IP])[:20]
485 return first[IP].__class__(header + buffer.getvalue())
487 header = bytes(first[Ether])[:34]
488 return first[Ether].__class__(header + buffer.getvalue())
491 def reassemble4_ether(listoffragments):
492 return reassemble4_core(listoffragments, False)
495 def reassemble4(listoffragments):
496 return reassemble4_core(listoffragments, True)
499 class UnexpectedPacketError(Exception):
500 def __init__(self, packet, msg=""):
505 return f"\nUnexpected packet:\n{pr(self.packet)}{self.msg}"
508 def recursive_dict_merge(dict_base, dict_update):
509 """Recursively merge base dict with update dict, return merged dict"""
510 for key in dict_update:
512 if type(dict_update[key]) is dict:
513 dict_base[key] = recursive_dict_merge(dict_base[key], dict_update[key])
515 dict_base[key] = dict_update[key]
517 dict_base[key] = dict_update[key]
521 class StatsDiff(UserDict):
523 Diff dictionary is a dictionary of dictionaries of interesting stats:
527 "err" : { '/error/counter1' : 4, },
528 sw_if_index1 : { '/stat/segment/counter1' : 5,
529 '/stat/segment/counter2' : 6,
531 sw_if_index2 : { '/stat/segment/counter1' : 7,
535 It describes a per sw-if-index diffset, where each key is stat segment
536 path and value is the expected change for that counter for sw-if-index.
537 Special case string "err" is used for error counters, which are not per
541 __slots__ = () # prevent setting properties to act like a dictionary
543 def __init__(self, data):
544 super().__init__(data)
546 def __or__(self, other):
547 return recursive_dict_merge(deepcopy(self.data), other)