""" test framework utilities """
+import abc
import socket
+from socket import AF_INET6
+import six
import sys
-from abc import abstractmethod, ABCMeta
-from cStringIO import StringIO
-from scapy.utils6 import in6_mactoifaceid
+import os.path
+import scapy.compat
from scapy.layers.l2 import Ether
-from scapy.packet import Raw
from scapy.layers.inet import IP
from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
IPv6ExtHdrHopByHop
+from scapy.packet import Raw
from scapy.utils import hexdump
-from socket import AF_INET6
+from scapy.utils6 import in6_mactoifaceid
+
+from io import BytesIO
+from vpp_papi import mac_pton
def ppp(headline, packet):
""" Return string containing the output of scapy packet.show() call. """
- o = StringIO()
- old_stdout = sys.stdout
- sys.stdout = o
- print(headline)
- hexdump(packet)
- print("")
- packet.show()
- sys.stdout = old_stdout
- return o.getvalue()
+ return '%s\n%s\n\n%s\n' % (headline,
+ hexdump(packet, dump=True),
+ packet.show(dump=True))
def ppc(headline, capture, limit=10):
for ip in ip4_range(ip4, s, e))
-def mactobinary(mac):
- """ Convert the : separated format into binary packet data for the API """
- return mac.replace(':', '').decode('hex')
-
-
+# wrapper around scapy library function.
def mk_ll_addr(mac):
- euid = in6_mactoifaceid(mac)
+ euid = in6_mactoifaceid(str(mac))
addr = "fe80::" + euid
return addr
socket.inet_pton(socket.AF_INET6, ip6))
-def check_core_path(logger, core_path):
+def get_core_path(tempdir):
+ return "%s/%s" % (tempdir, get_core_pattern())
+
+
+def is_core_present(tempdir):
+ return os.path.isfile(get_core_path(tempdir))
+
+
+def get_core_pattern():
with open("/proc/sys/kernel/core_pattern", "r") as f:
- corefmt = f.read()
- if corefmt.startswith("|"):
- logger.error(
- "WARNING: redirecting the core dump through a"
- " filter may result in truncated dumps.")
- logger.error(
- " You may want to check the filter settings"
- " or uninstall it and edit the"
- " /proc/sys/kernel/core_pattern accordingly.")
- logger.error(
- " current core pattern is: %s" % corefmt)
+ corefmt = f.read().strip()
+ return corefmt
+
+
+def check_core_path(logger, core_path):
+ corefmt = get_core_pattern()
+ if corefmt.startswith("|"):
+ logger.error(
+ "WARNING: redirecting the core dump through a"
+ " filter may result in truncated dumps.")
+ logger.error(
+ " You may want to check the filter settings"
+ " or uninstall it and edit the"
+ " /proc/sys/kernel/core_pattern accordingly.")
+ logger.error(
+ " current core pattern is: %s" % corefmt)
class NumericConstant(object):
- __metaclass__ = ABCMeta
desc_dict = {}
- @abstractmethod
def __init__(self, value):
self._value = value
@property
def bin_mac(self):
""" MAC address """
- return mactobinary(self._mac)
+ return mac_pton(self._mac)
@property
def ip4(self):
"""
logger = LoggerWrapper(_logger)
logger.debug(ppp("Fragmenting packet:", packet))
- packet = packet.__class__(str(packet)) # recalculate all values
+ packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
if len(packet[IP].options) > 0:
raise Exception("Not implemented")
if len(packet) <= fragsize:
pre_ip_len = len(packet) - len(packet[IP])
ip_header_len = packet[IP].ihl * 4
- hex_packet = str(packet)
+ hex_packet = scapy.compat.raw(packet)
hex_headers = hex_packet[:(pre_ip_len + ip_header_len)]
hex_payload = hex_packet[(pre_ip_len + ip_header_len):]
pkts = []
ihl = packet[IP].ihl
otl = len(packet[IP])
- nfb = (fragsize - pre_ip_len - ihl * 4) / 8
+ nfb = int((fragsize - pre_ip_len - ihl * 4) / 8)
fo = packet[IP].frag
p = packet.__class__(hex_headers + hex_payload[:nfb * 8])
:returns: list of fragments
"""
logger = LoggerWrapper(_logger)
- packet = packet.__class__(str(packet)) # recalculate all values
+ packet = packet.__class__(scapy.compat.raw(packet)) # recalc. all values
if len(packet) <= fragsize:
return [packet]
logger.debug(ppp("Fragmenting packet:", packet))
logger.debug(ppp("Per-fragment headers:", per_fragment_headers))
ext_and_upper_layer = packet.getlayer(last_per_fragment_hdr)[1]
- hex_payload = str(ext_and_upper_layer)
+ hex_payload = scapy.compat.raw(ext_and_upper_layer)
logger.debug("Payload length is %s" % len(hex_payload))
logger.debug(ppp("Ext and upper layer:", ext_and_upper_layer))
fragment_ext_hdr = IPv6ExtHdrFragment()
logger.debug(ppp("Fragment header:", fragment_ext_hdr))
+ len_ext_and_upper_layer_payload = len(ext_and_upper_layer.payload)
+ if not len_ext_and_upper_layer_payload and \
+ hasattr(ext_and_upper_layer, "data"):
+ len_ext_and_upper_layer_payload = len(ext_and_upper_layer.data)
+
if len(per_fragment_headers) + len(fragment_ext_hdr) +\
- len(ext_and_upper_layer) - len(ext_and_upper_layer.payload)\
+ len(ext_and_upper_layer) - len_ext_and_upper_layer_payload\
> fragsize:
raise Exception("Cannot fragment this packet - MTU too small "
"(%s, %s, %s, %s, %s)" % (
len(per_fragment_headers), len(fragment_ext_hdr),
len(ext_and_upper_layer),
- len(ext_and_upper_layer.payload), fragsize))
+ len_ext_and_upper_layer_payload, fragsize))
orig_nh = packet[IPv6].nh
p = per_fragment_headers
del p[IPv6].nh
p = p / fragment_ext_hdr
del p[IPv6ExtHdrFragment].nh
- first_payload_len_nfb = (fragsize - len(p)) / 8
+ first_payload_len_nfb = int((fragsize - len(p)) / 8)
p = p / Raw(hex_payload[:first_payload_len_nfb * 8])
del p[IPv6].plen
p[IPv6ExtHdrFragment].nh = orig_nh
p[IPv6ExtHdrFragment].id = identification
p[IPv6ExtHdrFragment].offset = 0
p[IPv6ExtHdrFragment].m = 1
- p = p.__class__(str(p))
+ p = p.__class__(scapy.compat.raw(p))
logger.debug(ppp("Fragment %s:" % len(pkts), p))
pkts.append(p)
offset = first_payload_len_nfb * 8
del p[IPv6].nh
p = p / fragment_ext_hdr
del p[IPv6ExtHdrFragment].nh
- l_nfb = (fragsize - len(p)) / 8
+ l_nfb = int((fragsize - len(p)) / 8)
p = p / Raw(hex_payload[offset:offset + l_nfb * 8])
p[IPv6ExtHdrFragment].nh = orig_nh
p[IPv6ExtHdrFragment].id = identification
- p[IPv6ExtHdrFragment].offset = offset / 8
+ p[IPv6ExtHdrFragment].offset = int(offset / 8)
p[IPv6ExtHdrFragment].m = 1
- p = p.__class__(str(p))
+ p = p.__class__(scapy.compat.raw(p))
logger.debug(ppp("Fragment %s:" % len(pkts), p))
pkts.append(p)
offset = offset + l_nfb * 8
pkts[-1][IPv6ExtHdrFragment].m = 0 # reset more-flags in last fragment
return pkts
+
+
+def reassemble4_core(listoffragments, return_ip):
+ buffer = BytesIO()
+ first = listoffragments[0]
+ buffer.seek(20)
+ for pkt in listoffragments:
+ buffer.seek(pkt[IP].frag*8)
+ buffer.write(bytes(pkt[IP].payload))
+ first.len = len(buffer.getvalue()) + 20
+ first.flags = 0
+ del(first.chksum)
+ if return_ip:
+ header = bytes(first[IP])[:20]
+ return first[IP].__class__(header + buffer.getvalue())
+ else:
+ header = bytes(first[Ether])[:34]
+ return first[Ether].__class__(header + buffer.getvalue())
+
+
+def reassemble4_ether(listoffragments):
+ return reassemble4_core(listoffragments, False)
+
+
+def reassemble4(listoffragments):
+ return reassemble4_core(listoffragments, True)
Code Review / vpp.git / blobdiff