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     desc_dict = {}
 134
 135     def __init__(self, value):
 136         self._value = value
 137
 138     def __int__(self):
 139         return self._value
 140
 141     def __long__(self):
 142         return self._value
 143
 144     def __str__(self):
 145         if self._value in self.desc_dict:
 146             return self.desc_dict[self._value]
 147         return ""
 148
 149
 150 class Host:
 151     """Generic test host "connected" to VPPs interface."""
 152
 153     @property
 154     def mac(self):
 155         """MAC address"""
 156         return self._mac
 157
 158     @property
 159     def bin_mac(self):
 160         """MAC address"""
 161         return mac_pton(self._mac)
 162
 163     @property
 164     def ip4(self):
 165         """IPv4 address - string"""
 166         return self._ip4
 167
 168     @property
 169     def ip4n(self):
 170         """IPv4 address of remote host - raw, suitable as API parameter."""
 171         return socket.inet_pton(socket.AF_INET, self._ip4)
 172
 173     @property
 174     def ip6(self):
 175         """IPv6 address - string"""
 176         return self._ip6
 177
 178     @property
 179     def ip6n(self):
 180         """IPv6 address of remote host - raw, suitable as API parameter."""
 181         return socket.inet_pton(socket.AF_INET6, self._ip6)
 182
 183     @property
 184     def ip6_ll(self):
 185         """IPv6 link-local address - string"""
 186         return self._ip6_ll
 187
 188     @property
 189     def ip6n_ll(self):
 190         """IPv6 link-local address of remote host -
 191         raw, suitable as API parameter."""
 192         return socket.inet_pton(socket.AF_INET6, self._ip6_ll)
 193
 194     def __eq__(self, other):
 195         if isinstance(other, Host):
 196             return (
 197                 self.mac == other.mac
 198                 and self.ip4 == other.ip4
 199                 and self.ip6 == other.ip6
 200                 and self.ip6_ll == other.ip6_ll
 201             )
 202         else:
 203             return False
 204
 205     def __ne__(self, other):
 206         return not self.__eq__(other)
 207
 208     def __repr__(self):
 209         return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (
 210             self.mac,
 211             self.ip4,
 212             self.ip6,
 213             self.ip6_ll,
 214         )
 215
 216     def __hash__(self):
 217         return hash(self.__repr__())
 218
 219     def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
 220         self._mac = mac
 221         self._ip4 = ip4
 222         self._ip6 = ip6
 223         self._ip6_ll = ip6_ll
 224
 225
 226 class L4_Conn:
 227     """L4 'connection' tied to two VPP interfaces"""
 228
 229     def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
 230         self.testcase = testcase
 231         self.ifs = [None, None]
 232         self.ifs[0] = if1
 233         self.ifs[1] = if2
 234         self.address_family = af
 235         self.l4proto = l4proto
 236         self.ports = [None, None]
 237         self.ports[0] = port1
 238         self.ports[1] = port2
 239         self
 240
 241     def pkt(self, side, l4args={}, payload="x"):
 242         is_ip6 = 1 if self.address_family == AF_INET6 else 0
 243         s0 = side
 244         s1 = 1 - side
 245         src_if = self.ifs[s0]
 246         dst_if = self.ifs[s1]
 247         layer_3 = [
 248             IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
 249             IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6),
 250         ]
 251         merged_l4args = {"sport": self.ports[s0], "dport": self.ports[s1]}
 252         merged_l4args.update(l4args)
 253         p = (
 254             Ether(dst=src_if.local_mac, src=src_if.remote_mac)
 255             / layer_3[is_ip6]
 256             / self.l4proto(**merged_l4args)
 257             / Raw(payload)
 258         )
 259         return p
 260
 261     def send(self, side, flags=None, payload=""):
 262         l4args = {}
 263         if flags is not None:
 264             l4args["flags"] = flags
 265         self.ifs[side].add_stream(self.pkt(side, l4args=l4args, payload=payload))
 266         self.ifs[1 - side].enable_capture()
 267         self.testcase.pg_start()
 268
 269     def recv(self, side):
 270         p = self.ifs[side].wait_for_packet(1)
 271         return p
 272
 273     def send_through(self, side, flags=None, payload=""):
 274         self.send(side, flags, payload)
 275         p = self.recv(1 - side)
 276         return p
 277
 278     def send_pingpong(self, side, flags1=None, flags2=None):
 279         p1 = self.send_through(side, flags1)
 280         p2 = self.send_through(1 - side, flags2)
 281         return [p1, p2]
 282
 283
 284 class L4_CONN_SIDE:
 285     L4_CONN_SIDE_ZERO = 0
 286     L4_CONN_SIDE_ONE = 1
 287
 288
 289 def fragment_rfc791(packet, fragsize, logger=null_logger):
 290     """
 291     Fragment an IPv4 packet per RFC 791
 292     :param packet: packet to fragment
 293     :param fragsize: size at which to fragment
 294     :note: IP options are not supported
 295     :returns: list of fragments
 296     """
 297     logger.debug(ppp("Fragmenting packet:", packet))
 298     packet = packet.__class__(scapy.compat.raw(packet))  # recalc. all values
 299     if len(packet[IP].options) > 0:
 300         raise Exception("Not implemented")
 301     if len(packet) <= fragsize:
 302         return [packet]
 303
 304     pre_ip_len = len(packet) - len(packet[IP])
 305     ip_header_len = packet[IP].ihl * 4
 306     hex_packet = scapy.compat.raw(packet)
 307     hex_headers = hex_packet[: (pre_ip_len + ip_header_len)]
 308     hex_payload = hex_packet[(pre_ip_len + ip_header_len) :]
 309
 310     pkts = []
 311     ihl = packet[IP].ihl
 312     otl = len(packet[IP])
 313     nfb = int((fragsize - pre_ip_len - ihl * 4) / 8)
 314     fo = packet[IP].frag
 315
 316     p = packet.__class__(hex_headers + hex_payload[: nfb * 8])
 317     p[IP].flags = "MF"
 318     p[IP].frag = fo
 319     p[IP].len = ihl * 4 + nfb * 8
 320     del p[IP].chksum
 321     pkts.append(p)
 322
 323     p = packet.__class__(hex_headers + hex_payload[nfb * 8 :])
 324     p[IP].len = otl - nfb * 8
 325     p[IP].frag = fo + nfb
 326     del p[IP].chksum
 327
 328     more_fragments = fragment_rfc791(p, fragsize, logger)
 329     pkts.extend(more_fragments)
 330
 331     return pkts
 332
 333
 334 def fragment_rfc8200(packet, identification, fragsize, logger=null_logger):
 335     """
 336     Fragment an IPv6 packet per RFC 8200
 337     :param packet: packet to fragment
 338     :param fragsize: size at which to fragment
 339     :note: IP options are not supported
 340     :returns: list of fragments
 341     """
 342     packet = packet.__class__(scapy.compat.raw(packet))  # recalc. all values
 343     if len(packet) <= fragsize:
 344         return [packet]
 345     logger.debug(ppp("Fragmenting packet:", packet))
 346     pkts = []
 347     counter = 0
 348     routing_hdr = None
 349     hop_by_hop_hdr = None
 350     upper_layer = None
 351     seen_ipv6 = False
 352     ipv6_nr = -1
 353     l = packet.getlayer(counter)
 354     while l is not None:
 355         if l.__class__ is IPv6:
 356             if seen_ipv6:
 357                 # ignore 2nd IPv6 header and everything below..
 358                 break
 359             ipv6_nr = counter
 360             seen_ipv6 = True
 361         elif l.__class__ is IPv6ExtHdrFragment:
 362             raise Exception("Already fragmented")
 363         elif l.__class__ is IPv6ExtHdrRouting:
 364             routing_hdr = counter
 365         elif l.__class__ is IPv6ExtHdrHopByHop:
 366             hop_by_hop_hdr = counter
 367         elif (
 368             seen_ipv6
 369             and not upper_layer
 370             and not l.__class__.__name__.startswith("IPv6ExtHdr")
 371         ):
 372             upper_layer = counter
 373         counter = counter + 1
 374         l = packet.getlayer(counter)
 375
 376     logger.debug(
 377         "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)"
 378         % (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer)
 379     )
 380
 381     if upper_layer is None:
 382         raise Exception("Upper layer header not found in IPv6 packet")
 383
 384     last_per_fragment_hdr = ipv6_nr
 385     if routing_hdr is None:
 386         if hop_by_hop_hdr is not None:
 387             last_per_fragment_hdr = hop_by_hop_hdr
 388     else:
 389         last_per_fragment_hdr = routing_hdr
 390     logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
 391
 392     per_fragment_headers = packet.copy()
 393     per_fragment_headers[last_per_fragment_hdr].remove_payload()
 394     logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
 395
 396     ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
 397     hex_payload = scapy.compat.raw(ext_and_upper_layer)
 398     logger.debug("Payload length is %s" % len(hex_payload))
 399     logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
 400
 401     fragment_ext_hdr = IPv6ExtHdrFragment()
 402     logger.debug(ppp("Fragment header:", fragment_ext_hdr))
 403
 404     len_ext_and_upper_layer_payload = len(ext_and_upper_layer.payload)
 405     if not len_ext_and_upper_layer_payload and hasattr(ext_and_upper_layer, "data"):
 406         len_ext_and_upper_layer_payload = len(ext_and_upper_layer.data)
 407
 408     if (
 409         len(per_fragment_headers)
 410         + len(fragment_ext_hdr)
 411         + len(ext_and_upper_layer)
 412         - len_ext_and_upper_layer_payload
 413         > fragsize
 414     ):
 415         raise Exception(
 416             "Cannot fragment this packet - MTU too small "
 417             "(%s, %s, %s, %s, %s)"
 418             % (
 419                 len(per_fragment_headers),
 420                 len(fragment_ext_hdr),
 421                 len(ext_and_upper_layer),
 422                 len_ext_and_upper_layer_payload,
 423                 fragsize,
 424             )
 425         )
 426
 427     orig_nh = packet[IPv6].nh
 428     p = per_fragment_headers
 429     del p[IPv6].plen
 430     del p[IPv6].nh
 431     p = p / fragment_ext_hdr
 432     del p[IPv6ExtHdrFragment].nh
 433     first_payload_len_nfb = int((fragsize - len(p)) / 8)
 434     p = p / Raw(hex_payload[: first_payload_len_nfb * 8])
 435     del p[IPv6].plen
 436     p[IPv6ExtHdrFragment].nh = orig_nh
 437     p[IPv6ExtHdrFragment].id = identification
 438     p[IPv6ExtHdrFragment].offset = 0
 439     p[IPv6ExtHdrFragment].m = 1
 440     p = p.__class__(scapy.compat.raw(p))
 441     logger.debug(ppp("Fragment %s:" % len(pkts), p))
 442     pkts.append(p)
 443     offset = first_payload_len_nfb * 8
 444     logger.debug("Offset after first fragment: %s" % offset)
 445     while len(hex_payload) > offset:
 446         p = per_fragment_headers
 447         del p[IPv6].plen
 448         del p[IPv6].nh
 449         p = p / fragment_ext_hdr
 450         del p[IPv6ExtHdrFragment].nh
 451         l_nfb = int((fragsize - len(p)) / 8)
 452         p = p / Raw(hex_payload[offset : offset + l_nfb * 8])
 453         p[IPv6ExtHdrFragment].nh = orig_nh
 454         p[IPv6ExtHdrFragment].id = identification
 455         p[IPv6ExtHdrFragment].offset = int(offset / 8)
 456         p[IPv6ExtHdrFragment].m = 1
 457         p = p.__class__(scapy.compat.raw(p))
 458         logger.debug(ppp("Fragment %s:" % len(pkts), p))
 459         pkts.append(p)
 460         offset = offset + l_nfb * 8
 461
 462     pkts[-1][IPv6ExtHdrFragment].m = 0  # reset more-flags in last fragment
 463
 464     return pkts
 465
 466
 467 def reassemble4_core(listoffragments, return_ip):
 468     buffer = BytesIO()
 469     first = listoffragments[0]
 470     buffer.seek(20)
 471     for pkt in listoffragments:
 472         buffer.seek(pkt[IP].frag * 8)
 473         buffer.write(bytes(pkt[IP].payload))
 474     first.len = len(buffer.getvalue()) + 20
 475     first.flags = 0
 476     del first.chksum
 477     if return_ip:
 478         header = bytes(first[IP])[:20]
 479         return first[IP].__class__(header + buffer.getvalue())
 480     else:
 481         header = bytes(first[Ether])[:34]
 482         return first[Ether].__class__(header + buffer.getvalue())
 483
 484
 485 def reassemble4_ether(listoffragments):
 486     return reassemble4_core(listoffragments, False)
 487
 488
 489 def reassemble4(listoffragments):
 490     return reassemble4_core(listoffragments, True)
 491
 492
 493 class UnexpectedPacketError(Exception):
 494     def __init__(self, packet, msg=""):
 495         self.packet = packet
 496         self.msg = msg
 497
 498     def __str__(self):
 499         return f"\nUnexpected packet:\n{pr(self.packet)}{self.msg}"
 500
 501
 502 def recursive_dict_merge(dict_base, dict_update):
 503     """Recursively merge base dict with update dict, return merged dict"""
 504     for key in dict_update:
 505         if key in dict_base:
 506             if type(dict_update[key]) is dict:
 507                 dict_base[key] = recursive_dict_merge(dict_base[key], dict_update[key])
 508             else:
 509                 dict_base[key] = dict_update[key]
 510         else:
 511             dict_base[key] = dict_update[key]
 512     return dict_base
 513
 514
 515 class StatsDiff(UserDict):
 516     """
 517     Diff dictionary is a dictionary of dictionaries of interesting stats:
 518
 519         diff_dictionary =
 520         {
 521             "err" : { '/error/counter1' : 4, },
 522             sw_if_index1 : { '/stat/segment/counter1' : 5,
 523                              '/stat/segment/counter2' : 6,
 524                            },
 525             sw_if_index2 : { '/stat/segment/counter1' : 7,
 526                            },
 527         }
 528
 529     It describes a per sw-if-index diffset, where each key is stat segment
 530     path and value is the expected change for that counter for sw-if-index.
 531     Special case string "err" is used for error counters, which are not per
 532     sw-if-index.
 533     """
 534
 535     __slots__ = ()  # prevent setting properties to act like a dictionary
 536
 537     def __init__(self, data):
 538         super().__init__(data)
 539
 540     def __or__(self, other):
 541         return recursive_dict_merge(deepcopy(self.data), other)