test/util.py

   1 """ test framework utilities """
   2
   3 import socket
   4 import sys
   5 from abc import abstractmethod, ABCMeta
   6 from cStringIO import StringIO
   7 from scapy.utils6 import in6_mactoifaceid
   8
   9 from scapy.layers.l2 import Ether
  10 from scapy.packet import Raw
  11 from scapy.layers.inet import IP
  12 from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
  13     IPv6ExtHdrHopByHop
  14 from socket import AF_INET6
  15
  16
  17 def ppp(headline, packet):
  18     """ Return string containing the output of scapy packet.show() call. """
  19     o = StringIO()
  20     old_stdout = sys.stdout
  21     sys.stdout = o
  22     print(headline)
  23     packet.show()
  24     sys.stdout = old_stdout
  25     return o.getvalue()
  26
  27
  28 def ppc(headline, capture, limit=10):
  29     """ Return string containing ppp() printout for a capture.
  30
  31     :param headline: printed as first line of output
  32     :param capture: packets to print
  33     :param limit: limit the print to # of packets
  34     """
  35     if not capture:
  36         return headline
  37     tail = ""
  38     if limit < len(capture):
  39         tail = "\nPrint limit reached, %s out of %s packets printed" % (
  40             limit, len(capture))
  41     body = "".join([ppp("Packet #%s:" % count, p)
  42                     for count, p in zip(range(0, limit), capture)])
  43     return "%s\n%s%s" % (headline, body, tail)
  44
  45
  46 def ip4_range(ip4, s, e):
  47     tmp = ip4.rsplit('.', 1)[0]
  48     return ("%s.%d" % (tmp, i) for i in range(s, e))
  49
  50
  51 def ip4n_range(ip4n, s, e):
  52     ip4 = socket.inet_ntop(socket.AF_INET, ip4n)
  53     return (socket.inet_pton(socket.AF_INET, ip)
  54             for ip in ip4_range(ip4, s, e))
  55
  56
  57 def mactobinary(mac):
  58     """ Convert the : separated format into binary packet data for the API """
  59     return mac.replace(':', '').decode('hex')
  60
  61
  62 def mk_ll_addr(mac):
  63     euid = in6_mactoifaceid(mac)
  64     addr = "fe80::" + euid
  65     return addr
  66
  67
  68 def ip6_normalize(ip6):
  69     return socket.inet_ntop(socket.AF_INET6,
  70                             socket.inet_pton(socket.AF_INET6, ip6))
  71
  72
  73 def check_core_path(logger, core_path):
  74     with open("/proc/sys/kernel/core_pattern", "r") as f:
  75         corefmt = f.read()
  76         if corefmt.startswith("|"):
  77             logger.error(
  78                 "WARNING: redirecting the core dump through a"
  79                 " filter may result in truncated dumps.")
  80             logger.error(
  81                 "   You may want to check the filter settings"
  82                 " or uninstall it and edit the"
  83                 " /proc/sys/kernel/core_pattern accordingly.")
  84             logger.error(
  85                 "   current core pattern is: %s" % corefmt)
  86
  87
  88 class NumericConstant(object):
  89     __metaclass__ = ABCMeta
  90
  91     desc_dict = {}
  92
  93     @abstractmethod
  94     def __init__(self, value):
  95         self._value = value
  96
  97     def __int__(self):
  98         return self._value
  99
 100     def __long__(self):
 101         return self._value
 102
 103     def __str__(self):
 104         if self._value in self.desc_dict:
 105             return self.desc_dict[self._value]
 106         return ""
 107
 108
 109 class Host(object):
 110     """ Generic test host "connected" to VPPs interface. """
 111
 112     @property
 113     def mac(self):
 114         """ MAC address """
 115         return self._mac
 116
 117     @property
 118     def bin_mac(self):
 119         """ MAC address """
 120         return mactobinary(self._mac)
 121
 122     @property
 123     def ip4(self):
 124         """ IPv4 address - string """
 125         return self._ip4
 126
 127     @property
 128     def ip4n(self):
 129         """ IPv4 address of remote host - raw, suitable as API parameter."""
 130         return socket.inet_pton(socket.AF_INET, self._ip4)
 131
 132     @property
 133     def ip6(self):
 134         """ IPv6 address - string """
 135         return self._ip6
 136
 137     @property
 138     def ip6n(self):
 139         """ IPv6 address of remote host - raw, suitable as API parameter."""
 140         return socket.inet_pton(socket.AF_INET6, self._ip6)
 141
 142     @property
 143     def ip6_ll(self):
 144         """ IPv6 link-local address - string """
 145         return self._ip6_ll
 146
 147     @property
 148     def ip6n_ll(self):
 149         """ IPv6 link-local address of remote host -
 150         raw, suitable as API parameter."""
 151         return socket.inet_pton(socket.AF_INET6, self._ip6_ll)
 152
 153     def __eq__(self, other):
 154         if isinstance(other, Host):
 155             return (self.mac == other.mac and
 156                     self.ip4 == other.ip4 and
 157                     self.ip6 == other.ip6 and
 158                     self.ip6_ll == other.ip6_ll)
 159         else:
 160             return False
 161
 162     def __ne__(self, other):
 163         return not self.__eq__(other)
 164
 165     def __repr__(self):
 166         return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (self.mac,
 167                                                             self.ip4,
 168                                                             self.ip6,
 169                                                             self.ip6_ll)
 170
 171     def __hash__(self):
 172         return hash(self.__repr__())
 173
 174     def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
 175         self._mac = mac
 176         self._ip4 = ip4
 177         self._ip6 = ip6
 178         self._ip6_ll = ip6_ll
 179
 180
 181 class ForeignAddressFactory(object):
 182     count = 0
 183     prefix_len = 24
 184     net_template = '10.10.10.{}'
 185     net = net_template.format(0) + '/' + str(prefix_len)
 186
 187     def get_ip4(self):
 188         if self.count > 255:
 189             raise Exception("Network host address exhaustion")
 190         self.count += 1
 191         return self.net_template.format(self.count)
 192
 193
 194 class L4_Conn():
 195     """ L4 'connection' tied to two VPP interfaces """
 196
 197     def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
 198         self.testcase = testcase
 199         self.ifs = [None, None]
 200         self.ifs[0] = if1
 201         self.ifs[1] = if2
 202         self.address_family = af
 203         self.l4proto = l4proto
 204         self.ports = [None, None]
 205         self.ports[0] = port1
 206         self.ports[1] = port2
 207         self
 208
 209     def pkt(self, side, l4args={}, payload="x"):
 210         is_ip6 = 1 if self.address_family == AF_INET6 else 0
 211         s0 = side
 212         s1 = 1 - side
 213         src_if = self.ifs[s0]
 214         dst_if = self.ifs[s1]
 215         layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
 216                    IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)]
 217         merged_l4args = {'sport': self.ports[s0], 'dport': self.ports[s1]}
 218         merged_l4args.update(l4args)
 219         p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 220              layer_3[is_ip6] /
 221              self.l4proto(**merged_l4args) /
 222              Raw(payload))
 223         return p
 224
 225     def send(self, side, flags=None, payload=""):
 226         l4args = {}
 227         if flags is not None:
 228             l4args['flags'] = flags
 229         self.ifs[side].add_stream(self.pkt(side,
 230                                            l4args=l4args, payload=payload))
 231         self.ifs[1 - side].enable_capture()
 232         self.testcase.pg_start()
 233
 234     def recv(self, side):
 235         p = self.ifs[side].wait_for_packet(1)
 236         return p
 237
 238     def send_through(self, side, flags=None, payload=""):
 239         self.send(side, flags, payload)
 240         p = self.recv(1 - side)
 241         return p
 242
 243     def send_pingpong(self, side, flags1=None, flags2=None):
 244         p1 = self.send_through(side, flags1)
 245         p2 = self.send_through(1 - side, flags2)
 246         return [p1, p2]
 247
 248
 249 class L4_CONN_SIDE:
 250     L4_CONN_SIDE_ZERO = 0
 251     L4_CONN_SIDE_ONE = 1
 252
 253
 254 class LoggerWrapper(object):
 255     def __init__(self, logger=None):
 256         self._logger = logger
 257
 258     def debug(self, *args, **kwargs):
 259         if self._logger:
 260             self._logger.debug(*args, **kwargs)
 261
 262     def error(self, *args, **kwargs):
 263         if self._logger:
 264             self._logger.error(*args, **kwargs)
 265
 266
 267 def fragment_rfc791(packet, fragsize, _logger=None):
 268     """
 269     Fragment an IPv4 packet per RFC 791
 270     :param packet: packet to fragment
 271     :param fragsize: size at which to fragment
 272     :note: IP options are not supported
 273     :returns: list of fragments
 274     """
 275     logger = LoggerWrapper(_logger)
 276     logger.debug(ppp("Fragmenting packet:", packet))
 277     packet = packet.__class__(str(packet))  # recalculate all values
 278     if len(packet[IP].options) > 0:
 279         raise Exception("Not implemented")
 280     if len(packet) <= fragsize:
 281         return [packet]
 282
 283     pre_ip_len = len(packet) - len(packet[IP])
 284     ip_header_len = packet[IP].ihl * 4
 285     hex_packet = str(packet)
 286     hex_headers = hex_packet[:(pre_ip_len + ip_header_len)]
 287     hex_payload = hex_packet[(pre_ip_len + ip_header_len):]
 288
 289     pkts = []
 290     ihl = packet[IP].ihl
 291     otl = len(packet[IP])
 292     nfb = (fragsize - pre_ip_len - ihl * 4) / 8
 293     fo = packet[IP].frag
 294
 295     p = packet.__class__(hex_headers + hex_payload[:nfb * 8])
 296     p[IP].flags = "MF"
 297     p[IP].frag = fo
 298     p[IP].len = ihl * 4 + nfb * 8
 299     del p[IP].chksum
 300     pkts.append(p)
 301
 302     p = packet.__class__(hex_headers + hex_payload[nfb * 8:])
 303     p[IP].len = otl - nfb * 8
 304     p[IP].frag = fo + nfb
 305     del p[IP].chksum
 306
 307     more_fragments = fragment_rfc791(p, fragsize, _logger)
 308     pkts.extend(more_fragments)
 309
 310     return pkts
 311
 312
 313 def fragment_rfc8200(packet, identification, fragsize, _logger=None):
 314     """
 315     Fragment an IPv6 packet per RFC 8200
 316     :param packet: packet to fragment
 317     :param fragsize: size at which to fragment
 318     :note: IP options are not supported
 319     :returns: list of fragments
 320     """
 321     logger = LoggerWrapper(_logger)
 322     packet = packet.__class__(str(packet))  # recalculate all values
 323     if len(packet) <= fragsize:
 324         return [packet]
 325     logger.debug(ppp("Fragmenting packet:", packet))
 326     pkts = []
 327     counter = 0
 328     routing_hdr = None
 329     hop_by_hop_hdr = None
 330     upper_layer = None
 331     seen_ipv6 = False
 332     ipv6_nr = -1
 333     l = packet.getlayer(counter)
 334     while l is not None:
 335         if l.__class__ is IPv6:
 336             if seen_ipv6:
 337                 # ignore 2nd IPv6 header and everything below..
 338                 break
 339             ipv6_nr = counter
 340             seen_ipv6 = True
 341         elif l.__class__ is IPv6ExtHdrFragment:
 342             raise Exception("Already fragmented")
 343         elif l.__class__ is IPv6ExtHdrRouting:
 344             routing_hdr = counter
 345         elif l.__class__ is IPv6ExtHdrHopByHop:
 346             hop_by_hop_hdr = counter
 347         elif seen_ipv6 and not upper_layer and \
 348                 not l.__class__.__name__.startswith('IPv6ExtHdr'):
 349             upper_layer = counter
 350         counter = counter + 1
 351         l = packet.getlayer(counter)
 352
 353     logger.debug(
 354         "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)" %
 355         (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer))
 356
 357     if upper_layer is None:
 358         raise Exception("Upper layer header not found in IPv6 packet")
 359
 360     last_per_fragment_hdr = ipv6_nr
 361     if routing_hdr is None:
 362         if hop_by_hop_hdr is not None:
 363             last_per_fragment_hdr = hop_by_hop_hdr
 364     else:
 365         last_per_fragment_hdr = routing_hdr
 366     logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
 367
 368     per_fragment_headers = packet.copy()
 369     per_fragment_headers[last_per_fragment_hdr].remove_payload()
 370     logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
 371
 372     ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
 373     hex_payload = str(ext_and_upper_layer)
 374     logger.debug("Payload length is %s" % len(hex_payload))
 375     logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
 376
 377     fragment_ext_hdr = IPv6ExtHdrFragment()
 378     logger.debug(ppp("Fragment header:", fragment_ext_hdr))
 379
 380     if len(per_fragment_headers) + len(fragment_ext_hdr) +\
 381             len(ext_and_upper_layer) - len(ext_and_upper_layer.payload)\
 382             > fragsize:
 383         raise Exception("Cannot fragment this packet - MTU too small "
 384                         "(%s, %s, %s, %s, %s)" % (
 385                             len(per_fragment_headers), len(fragment_ext_hdr),
 386                             len(ext_and_upper_layer),
 387                             len(ext_and_upper_layer.payload), fragsize))
 388
 389     orig_nh = packet[IPv6].nh
 390     p = per_fragment_headers
 391     del p[IPv6].plen
 392     del p[IPv6].nh
 393     p = p / fragment_ext_hdr
 394     del p[IPv6ExtHdrFragment].nh
 395     first_payload_len_nfb = (fragsize - len(p)) / 8
 396     p = p / Raw(hex_payload[:first_payload_len_nfb * 8])
 397     del p[IPv6].plen
 398     p[IPv6ExtHdrFragment].nh = orig_nh
 399     p[IPv6ExtHdrFragment].id = identification
 400     p[IPv6ExtHdrFragment].offset = 0
 401     p[IPv6ExtHdrFragment].m = 1
 402     p = p.__class__(str(p))
 403     logger.debug(ppp("Fragment %s:" % len(pkts), p))
 404     pkts.append(p)
 405     offset = first_payload_len_nfb * 8
 406     logger.debug("Offset after first fragment: %s" % offset)
 407     while len(hex_payload) > offset:
 408         p = per_fragment_headers
 409         del p[IPv6].plen
 410         del p[IPv6].nh
 411         p = p / fragment_ext_hdr
 412         del p[IPv6ExtHdrFragment].nh
 413         l_nfb = (fragsize - len(p)) / 8
 414         p = p / Raw(hex_payload[offset:offset + l_nfb * 8])
 415         p[IPv6ExtHdrFragment].nh = orig_nh
 416         p[IPv6ExtHdrFragment].id = identification
 417         p[IPv6ExtHdrFragment].offset = offset / 8
 418         p[IPv6ExtHdrFragment].m = 1
 419         p = p.__class__(str(p))
 420         logger.debug(ppp("Fragment %s:" % len(pkts), p))
 421         pkts.append(p)
 422         offset = offset + l_nfb * 8
 423
 424     pkts[-1][IPv6ExtHdrFragment].m = 0  # reset more-flags in last fragment
 425
 426     return pkts