from scapy.layers.l2 import Ether
from scapy.packet import Raw
-from scapy.layers.inet import IP, UDP, TCP
-from scapy.layers.inet6 import IPv6, ICMPv6Unknown, ICMPv6EchoRequest
-from scapy.packet import Packet
-from socket import inet_pton, AF_INET, AF_INET6
+from scapy.layers.inet import IP
+from scapy.layers.inet6 import IPv6, IPv6ExtHdrFragment, IPv6ExtHdrRouting,\
+ IPv6ExtHdrHopByHop
+from socket import AF_INET6
def ppp(headline, packet):
tail = ""
if limit < len(capture):
tail = "\nPrint limit reached, %s out of %s packets printed" % (
- len(capture), limit)
- limit = len(capture)
+ limit, len(capture))
body = "".join([ppp("Packet #%s:" % count, p)
for count, p in zip(range(0, limit), capture)])
return "%s\n%s%s" % (headline, body, tail)
class L4_Conn():
""" L4 'connection' tied to two VPP interfaces """
+
def __init__(self, testcase, if1, if2, af, l4proto, port1, port2):
self.testcase = testcase
self.ifs = [None, None]
def pkt(self, side, l4args={}, payload="x"):
is_ip6 = 1 if self.address_family == AF_INET6 else 0
s0 = side
- s1 = 1-side
+ s1 = 1 - side
src_if = self.ifs[s0]
dst_if = self.ifs[s1]
layer_3 = [IP(src=src_if.remote_ip4, dst=dst_if.remote_ip4),
l4args['flags'] = flags
self.ifs[side].add_stream(self.pkt(side,
l4args=l4args, payload=payload))
- self.ifs[1-side].enable_capture()
+ self.ifs[1 - side].enable_capture()
self.testcase.pg_start()
def recv(self, side):
def send_through(self, side, flags=None, payload=""):
self.send(side, flags, payload)
- p = self.recv(1-side)
+ p = self.recv(1 - side)
return p
def send_pingpong(self, side, flags1=None, flags2=None):
p1 = self.send_through(side, flags1)
- p2 = self.send_through(1-side, flags2)
+ p2 = self.send_through(1 - side, flags2)
return [p1, p2]
class L4_CONN_SIDE:
L4_CONN_SIDE_ZERO = 0
L4_CONN_SIDE_ONE = 1
+
+
+class LoggerWrapper(object):
+ def __init__(self, logger=None):
+ self._logger = logger
+
+ def debug(self, *args, **kwargs):
+ if self._logger:
+ self._logger.debug(*args, **kwargs)
+
+ def error(self, *args, **kwargs):
+ if self._logger:
+ self._logger.error(*args, **kwargs)
+
+
+def fragment_rfc791(packet, fragsize, _logger=None):
+ """
+ Fragment an IPv4 packet per RFC 791
+ :param packet: packet to fragment
+ :param fragsize: size at which to fragment
+ :note: IP options are not supported
+ :returns: list of fragments
+ """
+ logger = LoggerWrapper(_logger)
+ logger.debug(ppp("Fragmenting packet:", packet))
+ packet = packet.__class__(str(packet)) # recalculate all values
+ if len(packet[IP].options) > 0:
+ raise Exception("Not implemented")
+ if len(packet) <= fragsize:
+ return [packet]
+
+ pre_ip_len = len(packet) - len(packet[IP])
+ ip_header_len = packet[IP].ihl * 4
+ hex_packet = str(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
+ fo = packet[IP].frag
+
+ p = packet.__class__(hex_headers + hex_payload[:nfb * 8])
+ p[IP].flags = "MF"
+ p[IP].frag = fo
+ p[IP].len = ihl * 4 + nfb * 8
+ del p[IP].chksum
+ pkts.append(p)
+
+ p = packet.__class__(hex_headers + hex_payload[nfb * 8:])
+ p[IP].len = otl - nfb * 8
+ p[IP].frag = fo + nfb
+ del p[IP].chksum
+
+ more_fragments = fragment_rfc791(p, fragsize, _logger)
+ pkts.extend(more_fragments)
+
+ return pkts
+
+
+def fragment_rfc8200(packet, identification, fragsize, _logger=None):
+ """
+ Fragment an IPv6 packet per RFC 8200
+ :param packet: packet to fragment
+ :param fragsize: size at which to fragment
+ :note: IP options are not supported
+ :returns: list of fragments
+ """
+ logger = LoggerWrapper(_logger)
+ packet = packet.__class__(str(packet)) # recalculate all values
+ if len(packet) <= fragsize:
+ return [packet]
+ logger.debug(ppp("Fragmenting packet:", packet))
+ pkts = []
+ counter = 0
+ routing_hdr = None
+ hop_by_hop_hdr = None
+ upper_layer = None
+ seen_ipv6 = False
+ ipv6_nr = -1
+ l = packet.getlayer(counter)
+ while l is not None:
+ if l.__class__ is IPv6:
+ if seen_ipv6:
+ # ignore 2nd IPv6 header and everything below..
+ break
+ ipv6_nr = counter
+ seen_ipv6 = True
+ elif l.__class__ is IPv6ExtHdrFragment:
+ raise Exception("Already fragmented")
+ elif l.__class__ is IPv6ExtHdrRouting:
+ routing_hdr = counter
+ elif l.__class__ is IPv6ExtHdrHopByHop:
+ hop_by_hop_hdr = counter
+ elif seen_ipv6 and not upper_layer and \
+ not l.__class__.__name__.startswith('IPv6ExtHdr'):
+ upper_layer = counter
+ counter = counter + 1
+ l = packet.getlayer(counter)
+
+ logger.debug(
+ "Layers seen: IPv6(#%s), Routing(#%s), HopByHop(#%s), upper(#%s)" %
+ (ipv6_nr, routing_hdr, hop_by_hop_hdr, upper_layer))
+
+ if upper_layer is None:
+ raise Exception("Upper layer header not found in IPv6 packet")
+
+ last_per_fragment_hdr = ipv6_nr
+ if routing_hdr is None:
+ if hop_by_hop_hdr is not None:
+ last_per_fragment_hdr = hop_by_hop_hdr
+ else:
+ last_per_fragment_hdr = routing_hdr
+ logger.debug("Last per-fragment hdr is #%s" % (last_per_fragment_hdr))
+
+ per_fragment_headers = packet.copy()
+ per_fragment_headers[last_per_fragment_hdr].remove_payload()
+ 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)
+ 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))
+
+ if len(per_fragment_headers) + len(fragment_ext_hdr) +\
+ 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))
+
+ orig_nh = packet[IPv6].nh
+ p = per_fragment_headers
+ del p[IPv6].plen
+ del p[IPv6].nh
+ p = p / fragment_ext_hdr
+ del p[IPv6ExtHdrFragment].nh
+ first_payload_len_nfb = (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))
+ logger.debug(ppp("Fragment %s:" % len(pkts), p))
+ pkts.append(p)
+ offset = first_payload_len_nfb * 8
+ logger.debug("Offset after first fragment: %s" % offset)
+ while len(hex_payload) > offset:
+ p = per_fragment_headers
+ del p[IPv6].plen
+ del p[IPv6].nh
+ p = p / fragment_ext_hdr
+ del p[IPv6ExtHdrFragment].nh
+ l_nfb = (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].m = 1
+ p = p.__class__(str(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
Code Review / vpp.git / blobdiff