1 """ test framework utilities """
5 from abc import abstractmethod, ABCMeta
6 from cStringIO import StringIO
7 from scapy.utils6 import in6_mactoifaceid
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,\
14 from socket import AF_INET6
17 def ppp(headline, packet):
18 """ Return string containing the output of scapy packet.show() call. """
20 old_stdout = sys.stdout
24 sys.stdout = old_stdout
28 def ppc(headline, capture, limit=10):
29 """ Return string containing ppp() printout for a capture.
31 :param headline: printed as first line of output
32 :param capture: packets to print
33 :param limit: limit the print to # of packets
38 if limit < len(capture):
39 tail = "\nPrint limit reached, %s out of %s packets printed" % (
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)
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))
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))
58 """ Convert the : separated format into binary packet data for the API """
59 return mac.replace(':', '').decode('hex')
63 euid = in6_mactoifaceid(mac)
64 addr = "fe80::" + euid
68 def ip6_normalize(ip6):
69 return socket.inet_ntop(socket.AF_INET6,
70 socket.inet_pton(socket.AF_INET6, ip6))
73 class NumericConstant(object):
74 __metaclass__ = ABCMeta
79 def __init__(self, value):
89 if self._value in self.desc_dict:
90 return self.desc_dict[self._value]
95 """ Generic test host "connected" to VPPs interface. """
105 return mactobinary(self._mac)
109 """ IPv4 address - string """
114 """ IPv4 address of remote host - raw, suitable as API parameter."""
115 return socket.inet_pton(socket.AF_INET, self._ip4)
119 """ IPv6 address - string """
124 """ IPv6 address of remote host - raw, suitable as API parameter."""
125 return socket.inet_pton(socket.AF_INET6, self._ip6)
129 """ IPv6 link-local address - string """
134 """ IPv6 link-local address of remote host -
135 raw, suitable as API parameter."""
136 return socket.inet_pton(socket.AF_INET6, self._ip6_ll)
138 def __eq__(self, other):
139 if isinstance(other, Host):
140 return (self.mac == other.mac and
141 self.ip4 == other.ip4 and
142 self.ip6 == other.ip6 and
143 self.ip6_ll == other.ip6_ll)
147 def __ne__(self, other):
148 return not self.__eq__(other)
151 return "Host { mac:%s ip4:%s ip6:%s ip6_ll:%s }" % (self.mac,
157 return hash(self.__repr__())
159 def __init__(self, mac=None, ip4=None, ip6=None, ip6_ll=None):
163 self._ip6_ll = ip6_ll
166 class ForeignAddressFactory(object):
169 net_template = '10.10.10.{}'
170 net = net_template.format(0) + '/' + str(prefix_len)
174 raise Exception("Network host address exhaustion")
176 return self.net_template.format(self.count)
180 """ L4 'connection' tied to two VPP interfaces """
182 def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
183 self.testcase = testcase
184 self.ifs = [None, None]
187 self.address_family = af
188 self.l4proto = l4proto
189 self.ports = [None, None]
190 self.ports[0] = port1
191 self.ports[1] = port2
194 def pkt(self, side, l4args={}, payload="x"):
195 is_ip6 = 1 if self.address_family == AF_INET6 else 0
198 src_if = self.ifs[s0]
199 dst_if = self.ifs[s1]
200 layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
201 IPv6(src=src_if.remote_ip6, dst=dst_if.remote_ip6)]
202 merged_l4args = {'sport': self.ports[s0], 'dport': self.ports[s1]}
203 merged_l4args.update(l4args)
204 p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
206 self.l4proto(**merged_l4args) /
210 def send(self, side, flags=None, payload=""):
212 if flags is not None:
213 l4args['flags'] = flags
214 self.ifs[side].add_stream(self.pkt(side,
215 l4args=l4args, payload=payload))
216 self.ifs[1 - side].enable_capture()
217 self.testcase.pg_start()
219 def recv(self, side):
220 p = self.ifs[side].wait_for_packet(1)
223 def send_through(self, side, flags=None, payload=""):
224 self.send(side, flags, payload)
225 p = self.recv(1 - side)
228 def send_pingpong(self, side, flags1=None, flags2=None):
229 p1 = self.send_through(side, flags1)
230 p2 = self.send_through(1 - side, flags2)
235 L4_CONN_SIDE_ZERO = 0
239 class LoggerWrapper(object):
240 def __init__(self, logger=None):
241 self._logger = logger
243 def debug(self, *args, **kwargs):
245 self._logger.debug(*args, **kwargs)
247 def error(self, *args, **kwargs):
249 self._logger.error(*args, **kwargs)
252 def fragment_rfc791(packet, fragsize, _logger=None):
254 Fragment an IPv4 packet per RFC 791
255 :param packet: packet to fragment
256 :param fragsize: size at which to fragment
257 :note: IP options are not supported
258 :returns: list of fragments
260 logger = LoggerWrapper(_logger)
261 logger.debug(ppp("Fragmenting packet:", packet))
262 packet = packet.__class__(str(packet)) # recalculate all values
263 if len(packet[IP].options) > 0:
264 raise Exception("Not implemented")
265 if len(packet) <= fragsize:
268 pre_ip_len = len(packet) - len(packet[IP])
269 ip_header_len = packet[IP].ihl * 4
270 hex_packet = str(packet)
271 hex_headers = hex_packet[:(pre_ip_len + ip_header_len)]
272 hex_payload = hex_packet[(pre_ip_len + ip_header_len):]
276 otl = len(packet[IP])
277 nfb = (fragsize - pre_ip_len - ihl * 4) / 8
280 p = packet.__class__(hex_headers + hex_payload[:nfb * 8])
283 p[IP].len = ihl * 4 + nfb * 8
287 p = packet.__class__(hex_headers + hex_payload[nfb * 8:])
288 p[IP].len = otl - nfb * 8
289 p[IP].frag = fo + nfb
292 more_fragments = fragment_rfc791(p, fragsize, _logger)
293 pkts.extend(more_fragments)
298 def fragment_rfc8200(packet, identification, fragsize, _logger=None):
300 Fragment an IPv6 packet per RFC 8200
301 :param packet: packet to fragment
302 :param fragsize: size at which to fragment
303 :note: IP options are not supported
304 :returns: list of fragments
306 logger = LoggerWrapper(_logger)
307 packet = packet.__class__(str(packet)) # recalculate all values
308 if len(packet) <= fragsize:
310 logger.debug(ppp("Fragmenting packet:", packet))
314 hop_by_hop_hdr = None
318 l = packet.getlayer(counter)
320 if l.__class__ is IPv6:
322 # ignore 2nd IPv6 header and everything below..
326 elif l.__class__ is IPv6ExtHdrFragment:
327 raise Exception("Already fragmented")
328 elif l.__class__ is IPv6ExtHdrRouting:
329 routing_hdr = counter
330 elif l.__class__ is IPv6ExtHdrHopByHop:
331 hop_by_hop_hdr = counter
332 elif seen_ipv6 and not upper_layer and \
333 not l.__class__.__name__.startswith('IPv6ExtHdr'):
334 upper_layer = counter
335 counter = counter + 1
336 l = packet.getlayer(counter)
339 "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)" %
340 (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer))
342 if upper_layer is None:
343 raise Exception("Upper layer header not found in IPv6 packet")
345 last_per_fragment_hdr = ipv6_nr
346 if routing_hdr is None:
347 if hop_by_hop_hdr is not None:
348 last_per_fragment_hdr = hop_by_hop_hdr
350 last_per_fragment_hdr = routing_hdr
351 logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
353 per_fragment_headers = packet.copy()
354 per_fragment_headers[last_per_fragment_hdr].remove_payload()
355 logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
357 ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
358 hex_payload = str(ext_and_upper_layer)
359 logger.debug("Payload length is %s" % len(hex_payload))
360 logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
362 fragment_ext_hdr = IPv6ExtHdrFragment()
363 logger.debug(ppp("Fragment header:", fragment_ext_hdr))
365 if len(per_fragment_headers) + len(fragment_ext_hdr) +\
366 len(ext_and_upper_layer) - len(ext_and_upper_layer.payload)\
368 raise Exception("Cannot fragment this packet - MTU too small "
369 "(%s, %s, %s, %s, %s)" % (
370 len(per_fragment_headers), len(fragment_ext_hdr),
371 len(ext_and_upper_layer),
372 len(ext_and_upper_layer.payload), fragsize))
374 orig_nh = packet[IPv6].nh
375 p = per_fragment_headers
378 p = p / fragment_ext_hdr
379 del p[IPv6ExtHdrFragment].nh
380 first_payload_len_nfb = (fragsize - len(p)) / 8
381 p = p / Raw(hex_payload[:first_payload_len_nfb * 8])
383 p[IPv6ExtHdrFragment].nh = orig_nh
384 p[IPv6ExtHdrFragment].id = identification
385 p[IPv6ExtHdrFragment].offset = 0
386 p[IPv6ExtHdrFragment].m = 1
387 p = p.__class__(str(p))
388 logger.debug(ppp("Fragment %s:" % len(pkts), p))
390 offset = first_payload_len_nfb * 8
391 logger.debug("Offset after first fragment: %s" % offset)
392 while len(hex_payload) > offset:
393 p = per_fragment_headers
396 p = p / fragment_ext_hdr
397 del p[IPv6ExtHdrFragment].nh
398 l_nfb = (fragsize - len(p)) / 8
399 p = p / Raw(hex_payload[offset:offset + l_nfb * 8])
400 p[IPv6ExtHdrFragment].nh = orig_nh
401 p[IPv6ExtHdrFragment].id = identification
402 p[IPv6ExtHdrFragment].offset = offset / 8
403 p[IPv6ExtHdrFragment].m = 1
404 p = p.__class__(str(p))
405 logger.debug(ppp("Fragment %s:" % len(pkts), p))
407 offset = offset + l_nfb * 8
409 pkts[-1][IPv6ExtHdrFragment].m = 0 # reset more-flags in last fragment