test/util.py

   1 """ test framework utilities """
   2
   3 import ipaddress
   4 import logging
   5 import socket
   6 from socket import AF_INET6
   7 import os.path
   8 import platform
   9 from copy import deepcopy
  10 from collections import UserDict
  11
  12 import scapy.compat
  13 from scapy.layers.l2 import Ether
  14 from scapy.layers.inet import IP
  15 from scapy.layers.inet6 import (
  16     IPv6,
  17     IPv6ExtHdrFragment,
  18     IPv6ExtHdrRouting,
  19     IPv6ExtHdrHopByHop,
  20 )
  21 from scapy.packet import Raw
  22 from scapy.utils import hexdump
  23 from scapy.utils6 import in6_mactoifaceid
  24
  25 from io import BytesIO
  26 from vpp_papi import mac_pton
  27
  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())
  31
  32
  33 def pr(packet):
  34     return packet.__repr__()
  35
  36
  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" % (
  40         headline,
  41         hexdump(packet, dump=True),
  42         packet.show(dump=True),
  43     )
  44
  45
  46 def ppc(headline, capture, limit=10):
  47     """Return string containing ppp() printout for a capture.
  48
  49     :param headline: printed as first line of output
  50     :param capture: packets to print
  51     :param limit: limit the print to # of packets
  52     """
  53     if not capture:
  54         return headline
  55     tail = ""
  56     if limit < len(capture):
  57         tail = "\nPrint limit reached, %s out of %s packets printed" % (
  58             limit,
  59             len(capture),
  60         )
  61     body = "".join(
  62         [ppp("Packet #%s:" % count, p) for count, p in zip(range(0, limit), capture)]
  63     )
  64     return "%s\n%s%s" % (headline, body, tail)
  65
  66
  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))
  70
  71
  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.")
  76     ip_as_int = int(ip)
  77     if ip.version == 4:
  78         mcast_mac = "01:00:5e:%02x:%02x:%02x" % (
  79             (ip_as_int >> 16) & 0x7F,
  80             (ip_as_int >> 8) & 0xFF,
  81             ip_as_int & 0xFF,
  82         )
  83     else:
  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,
  88             ip_as_int & 0xFF,
  89         )
  90     return mcast_mac
  91
  92
  93 # wrapper around scapy library function.
  94 def mk_ll_addr(mac):
  95     euid = in6_mactoifaceid(str(mac))
  96     addr = "fe80::" + euid
  97     return addr
  98
  99
 100 def ip6_normalize(ip6):
 101     return socket.inet_ntop(socket.AF_INET6, socket.inet_pton(socket.AF_INET6, ip6))
 102
 103
 104 def get_core_path(tempdir):
 105     return "%s/%s" % (tempdir, get_core_pattern())
 106
 107
 108 def is_core_present(tempdir):
 109     return os.path.isfile(get_core_path(tempdir))
 110
 111
 112 def get_core_pattern():
 113     if platform.uname().system == "FreeBSD":
 114         import sysctl
 115
 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()
 120     return corefmt
 121
 122
 123 def check_core_path(logger, core_path):
 124     corefmt = get_core_pattern()
 125     if corefmt.startswith("|"):
 126         logger.error(
 127             "WARNING: redirecting the core dump through a"
 128             " filter may result in truncated dumps."
 129         )
 130         logger.error(
 131             "   You may want to check the filter settings"
 132             " or uninstall it and edit the"
 133             " /proc/sys/kernel/core_pattern accordingly."
 134         )
 135         logger.error("   current core pattern is: %s" % corefmt)
 136
 137
 138 class NumericConstant:
 139     desc_dict = {}
 140
 141     def __init__(self, value):
 142         self._value = value
 143
 144     def __int__(self):
 145         return self._value
 146
 147     def __long__(self):
 148         return self._value
 149
 150     def __str__(self):
 151         if self._value in self.desc_dict:
 152             return self.desc_dict[self._value]
 153         return ""
 154
 155
 156 class Host:
 157     """Generic test host "connected" to VPPs interface."""
 158
 159     @property
 160     def mac(self):
 161         """MAC address"""
 162         return self._mac
 163
 164     @property
 165     def bin_mac(self):
 166         """MAC address"""
 167         return mac_pton(self._mac)
 168
 169     @property
 170     def ip4(self):
 171         """IPv4 address - string"""
 172         return self._ip4
 173
 174     @property
 175     def ip4n(self):
 176         """IPv4 address of remote host - raw, suitable as API parameter."""
 177         return socket.inet_pton(socket.AF_INET, self._ip4)
 178
 179     @property
 180     def ip6(self):
 181         """IPv6 address - string"""
 182         return self._ip6
 183
 184     @property
 185     def ip6n(self):
 186         """IPv6 address of remote host - raw, suitable as API parameter."""
 187         return socket.inet_pton(socket.AF_INET6, self._ip6)
 188
 189     @property
 190     def ip6_ll(self):
 191         """IPv6 link-local address - string"""
 192         return self._ip6_ll
 193
 194     @property
 195     def ip6n_ll(self):
 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)
 199
 200     def __eq__(self, other):
 201         if isinstance(other, Host):
 202             return (
 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
 207             )
 208         else:
 209             return False
 210
 211     def __ne__(self, other):
 212         return not self.__eq__(other)
 213
 214     def __repr__(self):
 215         return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (
 216             self.mac,
 217             self.ip4,
 218             self.ip6,
 219             self.ip6_ll,
 220         )
 221
 222     def __hash__(self):
 223         return hash(self.__repr__())
 224
 225     def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
 226         self._mac = mac
 227         self._ip4 = ip4
 228         self._ip6 = ip6
 229         self._ip6_ll = ip6_ll
 230
 231
 232 class L4_Conn:
 233     """L4 'connection' tied to two VPP interfaces"""
 234
 235     def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
 236         self.testcase = testcase
 237         self.ifs = [None, None]
 238         self.ifs[0] = if1
 239         self.ifs[1] = if2
 240         self.address_family = af
 241         self.l4proto = l4proto
 242         self.ports = [None, None]
 243         self.ports[0] = port1
 244         self.ports[1] = port2
 245         self
 246
 247     def pkt(self, side, l4args={}, payload="x"):
 248         is_ip6 = 1 if self.address_family == AF_INET6 else 0
 249         s0 = side
 250         s1 = 1 - side
 251         src_if = self.ifs[s0]
 252         dst_if = self.ifs[s1]
 253         layer_3 = [
 254             IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
 255             IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6),
 256         ]
 257         merged_l4args = {"sport": self.ports[s0], "dport": self.ports[s1]}
 258         merged_l4args.update(l4args)
 259         p = (
 260             Ether(dst=src_if.local_mac, src=src_if.remote_mac)
 261             / layer_3[is_ip6]
 262             / self.l4proto(**merged_l4args)
 263             / Raw(payload)
 264         )
 265         return p
 266
 267     def send(self, side, flags=None, payload=""):
 268         l4args = {}
 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()
 274
 275     def recv(self, side):
 276         p = self.ifs[side].wait_for_packet(1)
 277         return p
 278
 279     def send_through(self, side, flags=None, payload=""):
 280         self.send(side, flags, payload)
 281         p = self.recv(1 - side)
 282         return p
 283
 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)
 287         return [p1, p2]
 288
 289
 290 class L4_CONN_SIDE:
 291     L4_CONN_SIDE_ZERO = 0
 292     L4_CONN_SIDE_ONE = 1
 293
 294
 295 def fragment_rfc791(packet, fragsize, logger=null_logger):
 296     """
 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
 302     """
 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:
 308         return [packet]
 309
 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) :]
 315
 316     pkts = []
 317     ihl = packet[IP].ihl
 318     otl = len(packet[IP])
 319     nfb = int((fragsize - pre_ip_len - ihl * 4) / 8)
 320     fo = packet[IP].frag
 321
 322     p = packet.__class__(hex_headers + hex_payload[: nfb * 8])
 323     p[IP].flags = "MF"
 324     p[IP].frag = fo
 325     p[IP].len = ihl * 4 + nfb * 8
 326     del p[IP].chksum
 327     pkts.append(p)
 328
 329     p = packet.__class__(hex_headers + hex_payload[nfb * 8 :])
 330     p[IP].len = otl - nfb * 8
 331     p[IP].frag = fo + nfb
 332     del p[IP].chksum
 333
 334     more_fragments = fragment_rfc791(p, fragsize, logger)
 335     pkts.extend(more_fragments)
 336
 337     return pkts
 338
 339
 340 def fragment_rfc8200(packet, identification, fragsize, logger=null_logger):
 341     """
 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
 347     """
 348     packet = packet.__class__(scapy.compat.raw(packet))  # recalc. all values
 349     if len(packet) <= fragsize:
 350         return [packet]
 351     logger.debug(ppp("Fragmenting packet:", packet))
 352     pkts = []
 353     counter = 0
 354     routing_hdr = None
 355     hop_by_hop_hdr = None
 356     upper_layer = None
 357     seen_ipv6 = False
 358     ipv6_nr = -1
 359     l = packet.getlayer(counter)
 360     while l is not None:
 361         if l.__class__ is IPv6:
 362             if seen_ipv6:
 363                 # ignore 2nd IPv6 header and everything below..
 364                 break
 365             ipv6_nr = counter
 366             seen_ipv6 = True
 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
 373         elif (
 374             seen_ipv6
 375             and not upper_layer
 376             and not l.__class__.__name__.startswith("IPv6ExtHdr")
 377         ):
 378             upper_layer = counter
 379         counter = counter + 1
 380         l = packet.getlayer(counter)
 381
 382     logger.debug(
 383         "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)"
 384         % (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer)
 385     )
 386
 387     if upper_layer is None:
 388         raise Exception("Upper layer header not found in IPv6 packet")
 389
 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
 394     else:
 395         last_per_fragment_hdr = routing_hdr
 396     logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
 397
 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))
 401
 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))
 406
 407     fragment_ext_hdr = IPv6ExtHdrFragment()
 408     logger.debug(ppp("Fragment header:", fragment_ext_hdr))
 409
 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)
 413
 414     if (
 415         len(per_fragment_headers)
 416         + len(fragment_ext_hdr)
 417         + len(ext_and_upper_layer)
 418         - len_ext_and_upper_layer_payload
 419         > fragsize
 420     ):
 421         raise Exception(
 422             "Cannot fragment this packet - MTU too small "
 423             "(%s, %s, %s, %s, %s)"
 424             % (
 425                 len(per_fragment_headers),
 426                 len(fragment_ext_hdr),
 427                 len(ext_and_upper_layer),
 428                 len_ext_and_upper_layer_payload,
 429                 fragsize,
 430             )
 431         )
 432
 433     orig_nh = packet[IPv6].nh
 434     p = per_fragment_headers
 435     del p[IPv6].plen
 436     del p[IPv6].nh
 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])
 441     del p[IPv6].plen
 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))
 448     pkts.append(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
 453         del p[IPv6].plen
 454         del p[IPv6].nh
 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))
 465         pkts.append(p)
 466         offset = offset + l_nfb * 8
 467
 468     pkts[-1][IPv6ExtHdrFragment].m = 0  # reset more-flags in last fragment
 469
 470     return pkts
 471
 472
 473 def reassemble4_core(listoffragments, return_ip):
 474     buffer = BytesIO()
 475     first = listoffragments[0]
 476     buffer.seek(20)
 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
 481     first.flags = 0
 482     del first.chksum
 483     if return_ip:
 484         header = bytes(first[IP])[:20]
 485         return first[IP].__class__(header + buffer.getvalue())
 486     else:
 487         header = bytes(first[Ether])[:34]
 488         return first[Ether].__class__(header + buffer.getvalue())
 489
 490
 491 def reassemble4_ether(listoffragments):
 492     return reassemble4_core(listoffragments, False)
 493
 494
 495 def reassemble4(listoffragments):
 496     return reassemble4_core(listoffragments, True)
 497
 498
 499 class UnexpectedPacketError(Exception):
 500     def __init__(self, packet, msg=""):
 501         self.packet = packet
 502         self.msg = msg
 503
 504     def __str__(self):
 505         return f"\nUnexpected packet:\n{pr(self.packet)}{self.msg}"
 506
 507
 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:
 511         if key in dict_base:
 512             if type(dict_update[key]) is dict:
 513                 dict_base[key] = recursive_dict_merge(dict_base[key], dict_update[key])
 514             else:
 515                 dict_base[key] = dict_update[key]
 516         else:
 517             dict_base[key] = dict_update[key]
 518     return dict_base
 519
 520
 521 class StatsDiff(UserDict):
 522     """
 523     Diff dictionary is a dictionary of dictionaries of interesting stats:
 524
 525         diff_dictionary =
 526         {
 527             "err" : { '/error/counter1' : 4, },
 528             sw_if_index1 : { '/stat/segment/counter1' : 5,
 529                              '/stat/segment/counter2' : 6,
 530                            },
 531             sw_if_index2 : { '/stat/segment/counter1' : 7,
 532                            },
 533         }
 534
 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
 538     sw-if-index.
 539     """
 540
 541     __slots__ = ()  # prevent setting properties to act like a dictionary
 542
 543     def __init__(self, data):
 544         super().__init__(data)
 545
 546     def __or__(self, other):
 547         return recursive_dict_merge(deepcopy(self.data), other)