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 from copy import deepcopy
   9 from collections import UserDict
  10
  11 import scapy.compat
  12 from scapy.layers.l2 import Ether
  13 from scapy.layers.inet import IP
  14 from scapy.layers.inet6 import (
  15     IPv6,
  16     IPv6ExtHdrFragment,
  17     IPv6ExtHdrRouting,
  18     IPv6ExtHdrHopByHop,
  19 )
  20 from scapy.packet import Raw
  21 from scapy.utils import hexdump
  22 from scapy.utils6 import in6_mactoifaceid
  23
  24 from io import BytesIO
  25 from vpp_papi import mac_pton
  26
  27 # Set up an empty logger for the testcase that can be overridden as necessary
  28 null_logger = logging.getLogger("VppTestCase.util")
  29 null_logger.addHandler(logging.NullHandler())
  30
  31
  32 def pr(packet):
  33     return packet.__repr__()
  34
  35
  36 def ppp(headline, packet):
  37     """Return string containing headline and output of scapy packet.show()"""
  38     return "%s\n%s\n\n%s\n" % (
  39         headline,
  40         hexdump(packet, dump=True),
  41         packet.show(dump=True),
  42     )
  43
  44
  45 def ppc(headline, capture, limit=10):
  46     """Return string containing ppp() printout for a capture.
  47
  48     :param headline: printed as first line of output
  49     :param capture: packets to print
  50     :param limit: limit the print to # of packets
  51     """
  52     if not capture:
  53         return headline
  54     tail = ""
  55     if limit < len(capture):
  56         tail = "\nPrint limit reached, %s out of %s packets printed" % (
  57             limit,
  58             len(capture),
  59         )
  60     body = "".join(
  61         [ppp("Packet #%s:" % count, p) for count, p in zip(range(0, limit), capture)]
  62     )
  63     return "%s\n%s%s" % (headline, body, tail)
  64
  65
  66 def ip4_range(ip4, s, e):
  67     tmp = ip4.rsplit(".", 1)[0]
  68     return ("%s.%d" % (tmp, i) for i in range(s, e))
  69
  70
  71 def mcast_ip_to_mac(ip):
  72     ip = ipaddress.ip_address(ip)
  73     if not ip.is_multicast:
  74         raise ValueError("Must be multicast address.")
  75     ip_as_int = int(ip)
  76     if ip.version == 4:
  77         mcast_mac = "01:00:5e:%02x:%02x:%02x" % (
  78             (ip_as_int >> 16) & 0x7F,
  79             (ip_as_int >> 8) & 0xFF,
  80             ip_as_int & 0xFF,
  81         )
  82     else:
  83         mcast_mac = "33:33:%02x:%02x:%02x:%02x" % (
  84             (ip_as_int >> 24) & 0xFF,
  85             (ip_as_int >> 16) & 0xFF,
  86             (ip_as_int >> 8) & 0xFF,
  87             ip_as_int & 0xFF,
  88         )
  89     return mcast_mac
  90
  91
  92 # wrapper around scapy library function.
  93 def mk_ll_addr(mac):
  94     euid = in6_mactoifaceid(str(mac))
  95     addr = "fe80::" + euid
  96     return addr
  97
  98
  99 def ip6_normalize(ip6):
 100     return socket.inet_ntop(socket.AF_INET6, socket.inet_pton(socket.AF_INET6, ip6))
 101
 102
 103 def get_core_path(tempdir):
 104     return "%s/%s" % (tempdir, get_core_pattern())
 105
 106
 107 def is_core_present(tempdir):
 108     return os.path.isfile(get_core_path(tempdir))
 109
 110
 111 def get_core_pattern():
 112     with open("/proc/sys/kernel/core_pattern", "r") as f:
 113         corefmt = f.read().strip()
 114     return corefmt
 115
 116
 117 def check_core_path(logger, core_path):
 118     corefmt = get_core_pattern()
 119     if corefmt.startswith("|"):
 120         logger.error(
 121             "WARNING: redirecting the core dump through a"
 122             " filter may result in truncated dumps."
 123         )
 124         logger.error(
 125             "   You may want to check the filter settings"
 126             " or uninstall it and edit the"
 127             " /proc/sys/kernel/core_pattern accordingly."
 128         )
 129         logger.error("   current core pattern is: %s" % corefmt)
 130
 131
 132 class NumericConstant:
 133
 134     desc_dict = {}
 135
 136     def __init__(self, value):
 137         self._value = value
 138
 139     def __int__(self):
 140         return self._value
 141
 142     def __long__(self):
 143         return self._value
 144
 145     def __str__(self):
 146         if self._value in self.desc_dict:
 147             return self.desc_dict[self._value]
 148         return ""
 149
 150
 151 class Host:
 152     """Generic test host "connected" to VPPs interface."""
 153
 154     @property
 155     def mac(self):
 156         """MAC address"""
 157         return self._mac
 158
 159     @property
 160     def bin_mac(self):
 161         """MAC address"""
 162         return mac_pton(self._mac)
 163
 164     @property
 165     def ip4(self):
 166         """IPv4 address - string"""
 167         return self._ip4
 168
 169     @property
 170     def ip4n(self):
 171         """IPv4 address of remote host - raw, suitable as API parameter."""
 172         return socket.inet_pton(socket.AF_INET, self._ip4)
 173
 174     @property
 175     def ip6(self):
 176         """IPv6 address - string"""
 177         return self._ip6
 178
 179     @property
 180     def ip6n(self):
 181         """IPv6 address of remote host - raw, suitable as API parameter."""
 182         return socket.inet_pton(socket.AF_INET6, self._ip6)
 183
 184     @property
 185     def ip6_ll(self):
 186         """IPv6 link-local address - string"""
 187         return self._ip6_ll
 188
 189     @property
 190     def ip6n_ll(self):
 191         """IPv6 link-local address of remote host -
 192         raw, suitable as API parameter."""
 193         return socket.inet_pton(socket.AF_INET6, self._ip6_ll)
 194
 195     def __eq__(self, other):
 196         if isinstance(other, Host):
 197             return (
 198                 self.mac == other.mac
 199                 and self.ip4 == other.ip4
 200                 and self.ip6 == other.ip6
 201                 and self.ip6_ll == other.ip6_ll
 202             )
 203         else:
 204             return False
 205
 206     def __ne__(self, other):
 207         return not self.__eq__(other)
 208
 209     def __repr__(self):
 210         return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (
 211             self.mac,
 212             self.ip4,
 213             self.ip6,
 214             self.ip6_ll,
 215         )
 216
 217     def __hash__(self):
 218         return hash(self.__repr__())
 219
 220     def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
 221         self._mac = mac
 222         self._ip4 = ip4
 223         self._ip6 = ip6
 224         self._ip6_ll = ip6_ll
 225
 226
 227 class L4_Conn:
 228     """L4 'connection' tied to two VPP interfaces"""
 229
 230     def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
 231         self.testcase = testcase
 232         self.ifs = [None, None]
 233         self.ifs[0] = if1
 234         self.ifs[1] = if2
 235         self.address_family = af
 236         self.l4proto = l4proto
 237         self.ports = [None, None]
 238         self.ports[0] = port1
 239         self.ports[1] = port2
 240         self
 241
 242     def pkt(self, side, l4args={}, payload="x"):
 243         is_ip6 = 1 if self.address_family == AF_INET6 else 0
 244         s0 = side
 245         s1 = 1 - side
 246         src_if = self.ifs[s0]
 247         dst_if = self.ifs[s1]
 248         layer_3 = [
 249             IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
 250             IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6),
 251         ]
 252         merged_l4args = {"sport": self.ports[s0], "dport": self.ports[s1]}
 253         merged_l4args.update(l4args)
 254         p = (
 255             Ether(dst=src_if.local_mac, src=src_if.remote_mac)
 256             / layer_3[is_ip6]
 257             / self.l4proto(**merged_l4args)
 258             / Raw(payload)
 259         )
 260         return p
 261
 262     def send(self, side, flags=None, payload=""):
 263         l4args = {}
 264         if flags is not None:
 265             l4args["flags"] = flags
 266         self.ifs[side].add_stream(self.pkt(side, l4args=l4args, payload=payload))
 267         self.ifs[1 - side].enable_capture()
 268         self.testcase.pg_start()
 269
 270     def recv(self, side):
 271         p = self.ifs[side].wait_for_packet(1)
 272         return p
 273
 274     def send_through(self, side, flags=None, payload=""):
 275         self.send(side, flags, payload)
 276         p = self.recv(1 - side)
 277         return p
 278
 279     def send_pingpong(self, side, flags1=None, flags2=None):
 280         p1 = self.send_through(side, flags1)
 281         p2 = self.send_through(1 - side, flags2)
 282         return [p1, p2]
 283
 284
 285 class L4_CONN_SIDE:
 286     L4_CONN_SIDE_ZERO = 0
 287     L4_CONN_SIDE_ONE = 1
 288
 289
 290 def fragment_rfc791(packet, fragsize, logger=null_logger):
 291     """
 292     Fragment an IPv4 packet per RFC 791
 293     :param packet: packet to fragment
 294     :param fragsize: size at which to fragment
 295     :note: IP options are not supported
 296     :returns: list of fragments
 297     """
 298     logger.debug(ppp("Fragmenting packet:", packet))
 299     packet = packet.__class__(scapy.compat.raw(packet))  # recalc. all values
 300     if len(packet[IP].options) > 0:
 301         raise Exception("Not implemented")
 302     if len(packet) <= fragsize:
 303         return [packet]
 304
 305     pre_ip_len = len(packet) - len(packet[IP])
 306     ip_header_len = packet[IP].ihl * 4
 307     hex_packet = scapy.compat.raw(packet)
 308     hex_headers = hex_packet[: (pre_ip_len + ip_header_len)]
 309     hex_payload = hex_packet[(pre_ip_len + ip_header_len) :]
 310
 311     pkts = []
 312     ihl = packet[IP].ihl
 313     otl = len(packet[IP])
 314     nfb = int((fragsize - pre_ip_len - ihl * 4) / 8)
 315     fo = packet[IP].frag
 316
 317     p = packet.__class__(hex_headers + hex_payload[: nfb * 8])
 318     p[IP].flags = "MF"
 319     p[IP].frag = fo
 320     p[IP].len = ihl * 4 + nfb * 8
 321     del p[IP].chksum
 322     pkts.append(p)
 323
 324     p = packet.__class__(hex_headers + hex_payload[nfb * 8 :])
 325     p[IP].len = otl - nfb * 8
 326     p[IP].frag = fo + nfb
 327     del p[IP].chksum
 328
 329     more_fragments = fragment_rfc791(p, fragsize, logger)
 330     pkts.extend(more_fragments)
 331
 332     return pkts
 333
 334
 335 def fragment_rfc8200(packet, identification, fragsize, logger=null_logger):
 336     """
 337     Fragment an IPv6 packet per RFC 8200
 338     :param packet: packet to fragment
 339     :param fragsize: size at which to fragment
 340     :note: IP options are not supported
 341     :returns: list of fragments
 342     """
 343     packet = packet.__class__(scapy.compat.raw(packet))  # recalc. all values
 344     if len(packet) <= fragsize:
 345         return [packet]
 346     logger.debug(ppp("Fragmenting packet:", packet))
 347     pkts = []
 348     counter = 0
 349     routing_hdr = None
 350     hop_by_hop_hdr = None
 351     upper_layer = None
 352     seen_ipv6 = False
 353     ipv6_nr = -1
 354     l = packet.getlayer(counter)
 355     while l is not None:
 356         if l.__class__ is IPv6:
 357             if seen_ipv6:
 358                 # ignore 2nd IPv6 header and everything below..
 359                 break
 360             ipv6_nr = counter
 361             seen_ipv6 = True
 362         elif l.__class__ is IPv6ExtHdrFragment:
 363             raise Exception("Already fragmented")
 364         elif l.__class__ is IPv6ExtHdrRouting:
 365             routing_hdr = counter
 366         elif l.__class__ is IPv6ExtHdrHopByHop:
 367             hop_by_hop_hdr = counter
 368         elif (
 369             seen_ipv6
 370             and not upper_layer
 371             and not l.__class__.__name__.startswith("IPv6ExtHdr")
 372         ):
 373             upper_layer = counter
 374         counter = counter + 1
 375         l = packet.getlayer(counter)
 376
 377     logger.debug(
 378         "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)"
 379         % (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer)
 380     )
 381
 382     if upper_layer is None:
 383         raise Exception("Upper layer header not found in IPv6 packet")
 384
 385     last_per_fragment_hdr = ipv6_nr
 386     if routing_hdr is None:
 387         if hop_by_hop_hdr is not None:
 388             last_per_fragment_hdr = hop_by_hop_hdr
 389     else:
 390         last_per_fragment_hdr = routing_hdr
 391     logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
 392
 393     per_fragment_headers = packet.copy()
 394     per_fragment_headers[last_per_fragment_hdr].remove_payload()
 395     logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
 396
 397     ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
 398     hex_payload = scapy.compat.raw(ext_and_upper_layer)
 399     logger.debug("Payload length is %s" % len(hex_payload))
 400     logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
 401
 402     fragment_ext_hdr = IPv6ExtHdrFragment()
 403     logger.debug(ppp("Fragment header:", fragment_ext_hdr))
 404
 405     len_ext_and_upper_layer_payload = len(ext_and_upper_layer.payload)
 406     if not len_ext_and_upper_layer_payload and hasattr(ext_and_upper_layer, "data"):
 407         len_ext_and_upper_layer_payload = len(ext_and_upper_layer.data)
 408
 409     if (
 410         len(per_fragment_headers)
 411         + len(fragment_ext_hdr)
 412         + len(ext_and_upper_layer)
 413         - len_ext_and_upper_layer_payload
 414         > fragsize
 415     ):
 416         raise Exception(
 417             "Cannot fragment this packet - MTU too small "
 418             "(%s, %s, %s, %s, %s)"
 419             % (
 420                 len(per_fragment_headers),
 421                 len(fragment_ext_hdr),
 422                 len(ext_and_upper_layer),
 423                 len_ext_and_upper_layer_payload,
 424                 fragsize,
 425             )
 426         )
 427
 428     orig_nh = packet[IPv6].nh
 429     p = per_fragment_headers
 430     del p[IPv6].plen
 431     del p[IPv6].nh
 432     p = p / fragment_ext_hdr
 433     del p[IPv6ExtHdrFragment].nh
 434     first_payload_len_nfb = int((fragsize - len(p)) / 8)
 435     p = p / Raw(hex_payload[: first_payload_len_nfb * 8])
 436     del p[IPv6].plen
 437     p[IPv6ExtHdrFragment].nh = orig_nh
 438     p[IPv6ExtHdrFragment].id = identification
 439     p[IPv6ExtHdrFragment].offset = 0
 440     p[IPv6ExtHdrFragment].m = 1
 441     p = p.__class__(scapy.compat.raw(p))
 442     logger.debug(ppp("Fragment %s:" % len(pkts), p))
 443     pkts.append(p)
 444     offset = first_payload_len_nfb * 8
 445     logger.debug("Offset after first fragment: %s" % offset)
 446     while len(hex_payload) > offset:
 447         p = per_fragment_headers
 448         del p[IPv6].plen
 449         del p[IPv6].nh
 450         p = p / fragment_ext_hdr
 451         del p[IPv6ExtHdrFragment].nh
 452         l_nfb = int((fragsize - len(p)) / 8)
 453         p = p / Raw(hex_payload[offset : offset + l_nfb * 8])
 454         p[IPv6ExtHdrFragment].nh = orig_nh
 455         p[IPv6ExtHdrFragment].id = identification
 456         p[IPv6ExtHdrFragment].offset = int(offset / 8)
 457         p[IPv6ExtHdrFragment].m = 1
 458         p = p.__class__(scapy.compat.raw(p))
 459         logger.debug(ppp("Fragment %s:" % len(pkts), p))
 460         pkts.append(p)
 461         offset = offset + l_nfb * 8
 462
 463     pkts[-1][IPv6ExtHdrFragment].m = 0  # reset more-flags in last fragment
 464
 465     return pkts
 466
 467
 468 def reassemble4_core(listoffragments, return_ip):
 469     buffer = BytesIO()
 470     first = listoffragments[0]
 471     buffer.seek(20)
 472     for pkt in listoffragments:
 473         buffer.seek(pkt[IP].frag * 8)
 474         buffer.write(bytes(pkt[IP].payload))
 475     first.len = len(buffer.getvalue()) + 20
 476     first.flags = 0
 477     del first.chksum
 478     if return_ip:
 479         header = bytes(first[IP])[:20]
 480         return first[IP].__class__(header + buffer.getvalue())
 481     else:
 482         header = bytes(first[Ether])[:34]
 483         return first[Ether].__class__(header + buffer.getvalue())
 484
 485
 486 def reassemble4_ether(listoffragments):
 487     return reassemble4_core(listoffragments, False)
 488
 489
 490 def reassemble4(listoffragments):
 491     return reassemble4_core(listoffragments, True)
 492
 493
 494 class UnexpectedPacketError(Exception):
 495     def __init__(self, packet, msg=""):
 496         self.packet = packet
 497         self.msg = msg
 498
 499     def __str__(self):
 500         return f"\nUnexpected packet:\n{pr(self.packet)}{self.msg}"
 501
 502
 503 def recursive_dict_merge(dict_base, dict_update):
 504     """Recursively merge base dict with update dict, return merged dict"""
 505     for key in dict_update:
 506         if key in dict_base:
 507             if type(dict_update[key]) is dict:
 508                 dict_base[key] = recursive_dict_merge(dict_base[key], dict_update[key])
 509             else:
 510                 dict_base[key] = dict_update[key]
 511         else:
 512             dict_base[key] = dict_update[key]
 513     return dict_base
 514
 515
 516 class StatsDiff(UserDict):
 517     """
 518     Diff dictionary is a dictionary of dictionaries of interesting stats:
 519
 520         diff_dictionary =
 521         {
 522             "err" : { '/error/counter1' : 4, },
 523             sw_if_index1 : { '/stat/segment/counter1' : 5,
 524                              '/stat/segment/counter2' : 6,
 525                            },
 526             sw_if_index2 : { '/stat/segment/counter1' : 7,
 527                            },
 528         }
 529
 530     It describes a per sw-if-index diffset, where each key is stat segment
 531     path and value is the expected change for that counter for sw-if-index.
 532     Special case string "err" is used for error counters, which are not per
 533     sw-if-index.
 534     """
 535
 536     __slots__ = ()  # prevent setting properties to act like a dictionary
 537
 538     def __init__(self, data):
 539         super().__init__(data)
 540
 541     def __or__(self, other):
 542         return recursive_dict_merge(deepcopy(self.data), other)