## This file is part of Scapy
|
## See http://www.secdev.org/projects/scapy for more informations
|
## Copyright (C) Philippe Biondi <phil@secdev.org>
|
## This program is published under a GPLv2 license
|
|
"""
|
Packet class. Binding mechanism. fuzz() method.
|
"""
|
|
from __future__ import absolute_import
|
from __future__ import print_function
|
import re
|
import time,itertools
|
import copy
|
import subprocess
|
|
from scapy.fields import StrField, ConditionalField, Emph, PacketListField, BitField, \
|
MultiEnumField, EnumField, FlagsField
|
from scapy.config import conf
|
from scapy.compat import *
|
from scapy.base_classes import BasePacket, Gen, SetGen, Packet_metaclass
|
from scapy.volatile import VolatileValue
|
from scapy.utils import import_hexcap,tex_escape,colgen,get_temp_file, \
|
ContextManagerSubprocess
|
from scapy.error import Scapy_Exception, log_runtime
|
from scapy.consts import PYX
|
import scapy.modules.six as six
|
|
try:
|
import pyx
|
except ImportError:
|
pass
|
|
|
class RawVal:
|
def __init__(self, val=""):
|
self.val = val
|
def __str__(self):
|
return str(self.val)
|
def __bytes__(self):
|
return raw(self.val)
|
def __repr__(self):
|
return "<RawVal [%r]>" % self.val
|
|
|
class Packet(six.with_metaclass(Packet_metaclass, BasePacket)):
|
__slots__ = [
|
"time", "sent_time", "name", "default_fields",
|
"overload_fields", "overloaded_fields", "fields", "fieldtype",
|
"packetfields",
|
"original", "explicit", "raw_packet_cache",
|
"raw_packet_cache_fields", "_pkt", "post_transforms",
|
# then payload and underlayer
|
"payload", "underlayer",
|
"name",
|
# used for sr()
|
"_answered",
|
# used when sniffing
|
"direction", "sniffed_on"
|
]
|
name = None
|
fields_desc = []
|
overload_fields = {}
|
payload_guess = []
|
show_indent = 1
|
show_summary = True
|
|
@classmethod
|
def from_hexcap(cls):
|
return cls(import_hexcap())
|
|
@classmethod
|
def upper_bonds(self):
|
for fval,upper in self.payload_guess:
|
print("%-20s %s" % (upper.__name__, ", ".join("%-12s" % ("%s=%r"%i) for i in six.iteritems(fval))))
|
|
@classmethod
|
def lower_bonds(self):
|
for lower,fval in six.iteritems(self._overload_fields):
|
print("%-20s %s" % (lower.__name__, ", ".join("%-12s" % ("%s=%r"%i) for i in six.iteritems(fval))))
|
|
def _unpickle(self, dlist):
|
"""Used to unpack pickling"""
|
self.__init__(b"".join(dlist))
|
return self
|
|
def __reduce__(self):
|
"""Used by pickling methods"""
|
return (self.__class__, (), (self.build(),))
|
|
def __reduce_ex__(self, proto):
|
"""Used by pickling methods"""
|
return self.__reduce__()
|
|
def __getstate__(self):
|
"""Mark object as pickable"""
|
return self.__reduce__()[2]
|
|
def __setstate__(self, state):
|
"""Rebuild state using pickable methods"""
|
return self._unpickle(state)
|
|
def __deepcopy__(self, memo):
|
"""Used by copy.deepcopy"""
|
return self.copy()
|
|
def __init__(self, _pkt=b"", post_transform=None, _internal=0, _underlayer=None, **fields):
|
self.time = time.time()
|
self.sent_time = None
|
self.name = (self.__class__.__name__
|
if self._name is None else
|
self._name)
|
self.default_fields = {}
|
self.overload_fields = self._overload_fields
|
self.overloaded_fields = {}
|
self.fields = {}
|
self.fieldtype = {}
|
self.packetfields = []
|
self.payload = NoPayload()
|
self.init_fields()
|
self.underlayer = _underlayer
|
self.original = _pkt
|
self.explicit = 0
|
self.raw_packet_cache = None
|
self.raw_packet_cache_fields = None
|
if _pkt:
|
self.dissect(_pkt)
|
if not _internal:
|
self.dissection_done(self)
|
for f, v in six.iteritems(fields):
|
self.fields[f] = self.get_field(f).any2i(self, v)
|
if isinstance(post_transform, list):
|
self.post_transforms = post_transform
|
elif post_transform is None:
|
self.post_transforms = []
|
else:
|
self.post_transforms = [post_transform]
|
|
def init_fields(self):
|
"""
|
Initialize each fields of the fields_desc dict
|
"""
|
self.do_init_fields(self.fields_desc)
|
|
def do_init_fields(self, flist):
|
"""
|
Initialize each fields of the fields_desc dict
|
"""
|
for f in flist:
|
self.default_fields[f.name] = copy.deepcopy(f.default)
|
self.fieldtype[f.name] = f
|
if f.holds_packets:
|
self.packetfields.append(f)
|
|
def dissection_done(self,pkt):
|
"""DEV: will be called after a dissection is completed"""
|
self.post_dissection(pkt)
|
self.payload.dissection_done(pkt)
|
|
def post_dissection(self, pkt):
|
"""DEV: is called after the dissection of the whole packet"""
|
pass
|
|
def get_field(self, fld):
|
"""DEV: returns the field instance from the name of the field"""
|
return self.fieldtype[fld]
|
|
def add_payload(self, payload):
|
if payload is None:
|
return
|
elif not isinstance(self.payload, NoPayload):
|
self.payload.add_payload(payload)
|
else:
|
if isinstance(payload, Packet):
|
self.payload = payload
|
payload.add_underlayer(self)
|
for t in self.aliastypes:
|
if t in payload.overload_fields:
|
self.overloaded_fields = payload.overload_fields[t]
|
break
|
elif isinstance(payload, bytes):
|
self.payload = conf.raw_layer(load=payload)
|
else:
|
raise TypeError("payload must be either 'Packet' or 'bytes', not [%s]" % repr(payload))
|
def remove_payload(self):
|
self.payload.remove_underlayer(self)
|
self.payload = NoPayload()
|
self.overloaded_fields = {}
|
def add_underlayer(self, underlayer):
|
self.underlayer = underlayer
|
def remove_underlayer(self,other):
|
self.underlayer = None
|
def copy(self):
|
"""Returns a deep copy of the instance."""
|
clone = self.__class__()
|
clone.fields = self.copy_fields_dict(self.fields)
|
clone.default_fields = self.copy_fields_dict(self.default_fields)
|
clone.overloaded_fields = self.overloaded_fields.copy()
|
clone.underlayer = self.underlayer
|
clone.explicit = self.explicit
|
clone.raw_packet_cache = self.raw_packet_cache
|
clone.raw_packet_cache_fields = self.copy_fields_dict(
|
self.raw_packet_cache_fields
|
)
|
clone.post_transforms = self.post_transforms[:]
|
clone.payload = self.payload.copy()
|
clone.payload.add_underlayer(clone)
|
clone.time = self.time
|
return clone
|
|
def getfieldval(self, attr):
|
if attr in self.fields:
|
return self.fields[attr]
|
if attr in self.overloaded_fields:
|
return self.overloaded_fields[attr]
|
if attr in self.default_fields:
|
return self.default_fields[attr]
|
return self.payload.getfieldval(attr)
|
|
def getfield_and_val(self, attr):
|
if attr in self.fields:
|
return self.get_field(attr),self.fields[attr]
|
if attr in self.overloaded_fields:
|
return self.get_field(attr),self.overloaded_fields[attr]
|
if attr in self.default_fields:
|
return self.get_field(attr),self.default_fields[attr]
|
|
def __getattr__(self, attr):
|
try:
|
fld, v = self.getfield_and_val(attr)
|
except TypeError:
|
return self.payload.__getattr__(attr)
|
if fld is not None:
|
return fld.i2h(self, v)
|
return v
|
|
def setfieldval(self, attr, val):
|
if attr in self.default_fields:
|
fld = self.get_field(attr)
|
if fld is None:
|
any2i = lambda x,y: y
|
else:
|
any2i = fld.any2i
|
self.fields[attr] = any2i(self, val)
|
self.explicit = 0
|
self.raw_packet_cache = None
|
self.raw_packet_cache_fields = None
|
elif attr == "payload":
|
self.remove_payload()
|
self.add_payload(val)
|
else:
|
self.payload.setfieldval(attr,val)
|
|
def __setattr__(self, attr, val):
|
if attr in self.__all_slots__:
|
return object.__setattr__(self, attr, val)
|
try:
|
return self.setfieldval(attr,val)
|
except AttributeError:
|
pass
|
return object.__setattr__(self, attr, val)
|
|
def delfieldval(self, attr):
|
if attr in self.fields:
|
del(self.fields[attr])
|
self.explicit = 0 # in case a default value must be explicited
|
self.raw_packet_cache = None
|
self.raw_packet_cache_fields = None
|
elif attr in self.default_fields:
|
pass
|
elif attr == "payload":
|
self.remove_payload()
|
else:
|
self.payload.delfieldval(attr)
|
|
def __delattr__(self, attr):
|
if attr == "payload":
|
return self.remove_payload()
|
if attr in self.__all_slots__:
|
return object.__delattr__(self, attr)
|
try:
|
return self.delfieldval(attr)
|
except AttributeError:
|
pass
|
return object.__delattr__(self, attr)
|
|
def _superdir(self):
|
"""
|
Return a list of slots and methods, including those from subclasses.
|
"""
|
attrs = set()
|
cls = self.__class__
|
if hasattr(cls, '__all_slots__'):
|
attrs.update(cls.__all_slots__)
|
for bcls in cls.__mro__:
|
if hasattr(bcls, '__dict__'):
|
attrs.update(bcls.__dict__)
|
return attrs
|
|
def __dir__(self):
|
"""
|
Add fields to tab completion list.
|
"""
|
return sorted(itertools.chain(self._superdir(), self.default_fields))
|
|
def __repr__(self):
|
s = ""
|
ct = conf.color_theme
|
for f in self.fields_desc:
|
if isinstance(f, ConditionalField) and not f._evalcond(self):
|
continue
|
if f.name in self.fields:
|
val = f.i2repr(self, self.fields[f.name])
|
elif f.name in self.overloaded_fields:
|
val = f.i2repr(self, self.overloaded_fields[f.name])
|
else:
|
continue
|
if isinstance(f, Emph) or f in conf.emph:
|
ncol = ct.emph_field_name
|
vcol = ct.emph_field_value
|
else:
|
ncol = ct.field_name
|
vcol = ct.field_value
|
|
|
s += " %s%s%s" % (ncol(f.name),
|
ct.punct("="),
|
vcol(val))
|
return "%s%s %s %s%s%s"% (ct.punct("<"),
|
ct.layer_name(self.__class__.__name__),
|
s,
|
ct.punct("|"),
|
repr(self.payload),
|
ct.punct(">"))
|
def __str__(self):
|
return str(self.build())
|
def __bytes__(self):
|
return self.build()
|
def __div__(self, other):
|
if isinstance(other, Packet):
|
cloneA = self.copy()
|
cloneB = other.copy()
|
cloneA.add_payload(cloneB)
|
return cloneA
|
elif isinstance(other, (bytes, str)):
|
return self/conf.raw_layer(load=other)
|
else:
|
return other.__rdiv__(self)
|
__truediv__ = __div__
|
def __rdiv__(self, other):
|
if isinstance(other, (bytes, str)):
|
return conf.raw_layer(load=other)/self
|
else:
|
raise TypeError
|
__rtruediv__ = __rdiv__
|
def __mul__(self, other):
|
if isinstance(other, int):
|
return [self]*other
|
else:
|
raise TypeError
|
def __rmul__(self,other):
|
return self.__mul__(other)
|
|
def __nonzero__(self):
|
return True
|
__bool__ = __nonzero__
|
def __len__(self):
|
return len(self.__bytes__())
|
def copy_field_value(self, fieldname, value):
|
return self.get_field(fieldname).do_copy(value)
|
def copy_fields_dict(self, fields):
|
if fields is None:
|
return None
|
return {fname: self.copy_field_value(fname, fval)
|
for fname, fval in six.iteritems(fields)}
|
def self_build(self, field_pos_list=None):
|
"""
|
Create the default layer regarding fields_desc dict
|
|
:param field_pos_list:
|
"""
|
if self.raw_packet_cache is not None:
|
for fname, fval in six.iteritems(self.raw_packet_cache_fields):
|
if self.getfieldval(fname) != fval:
|
self.raw_packet_cache = None
|
self.raw_packet_cache_fields = None
|
break
|
if self.raw_packet_cache is not None:
|
return self.raw_packet_cache
|
p=b""
|
for f in self.fields_desc:
|
val = self.getfieldval(f.name)
|
if isinstance(val, RawVal):
|
sval = raw(val)
|
p += sval
|
if field_pos_list is not None:
|
field_pos_list.append( (f.name, sval.encode("string_escape"), len(p), len(sval) ) )
|
else:
|
p = f.addfield(self, p, val)
|
return p
|
|
def do_build_payload(self):
|
"""
|
Create the default version of the payload layer
|
|
:return: a string of payload layer
|
"""
|
return self.payload.do_build()
|
|
def do_build(self):
|
"""
|
Create the default version of the layer
|
|
:return: a string of the packet with the payload
|
"""
|
if not self.explicit:
|
self = next(iter(self))
|
pkt = self.self_build()
|
for t in self.post_transforms:
|
pkt = t(pkt)
|
pay = self.do_build_payload()
|
if self.raw_packet_cache is None:
|
return self.post_build(pkt, pay)
|
else:
|
return pkt + pay
|
|
def build_padding(self):
|
return self.payload.build_padding()
|
|
def build(self):
|
"""
|
Create the current layer
|
|
:return: string of the packet with the payload
|
"""
|
p = self.do_build()
|
p += self.build_padding()
|
p = self.build_done(p)
|
return p
|
|
def post_build(self, pkt, pay):
|
"""
|
DEV: called right after the current layer is build.
|
|
:param str pkt: the current packet (build by self_buil function)
|
:param str pay: the packet payload (build by do_build_payload function)
|
:return: a string of the packet with the payload
|
"""
|
return pkt+pay
|
|
def build_done(self, p):
|
return self.payload.build_done(p)
|
|
def do_build_ps(self):
|
p = b""
|
pl = []
|
q = b""
|
for f in self.fields_desc:
|
if isinstance(f, ConditionalField) and not f._evalcond(self):
|
continue
|
p = f.addfield(self, p, self.getfieldval(f.name) )
|
if isinstance(p, bytes):
|
r = p[len(q):]
|
q = p
|
else:
|
r = b""
|
pl.append( (f, f.i2repr(self,self.getfieldval(f.name)), r) )
|
|
pkt,lst = self.payload.build_ps(internal=1)
|
p += pkt
|
lst.append( (self, pl) )
|
|
return p,lst
|
|
def build_ps(self,internal=0):
|
p,lst = self.do_build_ps()
|
# if not internal:
|
# pkt = self
|
# while pkt.haslayer(conf.padding_layer):
|
# pkt = pkt.getlayer(conf.padding_layer)
|
# lst.append( (pkt, [ ("loakjkjd", pkt.load, pkt.load) ] ) )
|
# p += pkt.load
|
# pkt = pkt.payload
|
return p,lst
|
|
|
def psdump(self, filename=None, **kargs):
|
"""
|
psdump(filename=None, layer_shift=0, rebuild=1)
|
|
Creates an EPS file describing a packet. If filename is not provided a
|
temporary file is created and gs is called.
|
|
:param filename: the file's filename
|
"""
|
canvas = self.canvas_dump(**kargs)
|
if filename is None:
|
fname = get_temp_file(autoext=".eps")
|
canvas.writeEPSfile(fname)
|
with ContextManagerSubprocess("psdump()", conf.prog.psreader):
|
subprocess.Popen([conf.prog.psreader, fname])
|
else:
|
canvas.writeEPSfile(filename)
|
|
def pdfdump(self, filename=None, **kargs):
|
"""
|
pdfdump(filename=None, layer_shift=0, rebuild=1)
|
|
Creates a PDF file describing a packet. If filename is not provided a
|
temporary file is created and xpdf is called.
|
|
:param filename: the file's filename
|
"""
|
canvas = self.canvas_dump(**kargs)
|
if filename is None:
|
fname = get_temp_file(autoext=".pdf")
|
canvas.writePDFfile(fname)
|
with ContextManagerSubprocess("pdfdump()", conf.prog.pdfreader):
|
subprocess.Popen([conf.prog.pdfreader, fname])
|
else:
|
canvas.writePDFfile(filename)
|
|
|
def canvas_dump(self, layer_shift=0, rebuild=1):
|
if PYX == 0:
|
raise ImportError("PyX and its depedencies must be installed")
|
canvas = pyx.canvas.canvas()
|
if rebuild:
|
p,t = self.__class__(raw(self)).build_ps()
|
else:
|
p,t = self.build_ps()
|
YTXT=len(t)
|
for n,l in t:
|
YTXT += len(l)
|
YTXT = float(YTXT)
|
YDUMP=YTXT
|
|
XSTART = 1
|
XDSTART = 10
|
y = 0.0
|
yd = 0.0
|
xd = 0
|
XMUL= 0.55
|
YMUL = 0.4
|
|
backcolor=colgen(0.6, 0.8, 1.0, trans=pyx.color.rgb)
|
forecolor=colgen(0.2, 0.5, 0.8, trans=pyx.color.rgb)
|
# backcolor=makecol(0.376, 0.729, 0.525, 1.0)
|
|
|
def hexstr(x):
|
s = []
|
for c in x:
|
s.append("%02x" % orb(c))
|
return " ".join(s)
|
|
|
def make_dump_txt(x,y,txt):
|
return pyx.text.text(XDSTART+x*XMUL, (YDUMP-y)*YMUL, r"\tt{%s}"%hexstr(txt), [pyx.text.size.Large])
|
|
def make_box(o):
|
return pyx.box.rect(o.left(), o.bottom(), o.width(), o.height(), relcenter=(0.5,0.5))
|
|
def make_frame(lst):
|
if len(lst) == 1:
|
b = lst[0].bbox()
|
b.enlarge(pyx.unit.u_pt)
|
return b.path()
|
else:
|
fb = lst[0].bbox()
|
fb.enlarge(pyx.unit.u_pt)
|
lb = lst[-1].bbox()
|
lb.enlarge(pyx.unit.u_pt)
|
if len(lst) == 2 and fb.left() > lb.right():
|
return pyx.path.path(pyx.path.moveto(fb.right(), fb.top()),
|
pyx.path.lineto(fb.left(), fb.top()),
|
pyx.path.lineto(fb.left(), fb.bottom()),
|
pyx.path.lineto(fb.right(), fb.bottom()),
|
pyx.path.moveto(lb.left(), lb.top()),
|
pyx.path.lineto(lb.right(), lb.top()),
|
pyx.path.lineto(lb.right(), lb.bottom()),
|
pyx.path.lineto(lb.left(), lb.bottom()))
|
else:
|
# XXX
|
gb = lst[1].bbox()
|
if gb != lb:
|
gb.enlarge(pyx.unit.u_pt)
|
kb = lst[-2].bbox()
|
if kb != gb and kb != lb:
|
kb.enlarge(pyx.unit.u_pt)
|
return pyx.path.path(pyx.path.moveto(fb.left(), fb.top()),
|
pyx.path.lineto(fb.right(), fb.top()),
|
pyx.path.lineto(fb.right(), kb.bottom()),
|
pyx.path.lineto(lb.right(), kb.bottom()),
|
pyx.path.lineto(lb.right(), lb.bottom()),
|
pyx.path.lineto(lb.left(), lb.bottom()),
|
pyx.path.lineto(lb.left(), gb.top()),
|
pyx.path.lineto(fb.left(), gb.top()),
|
pyx.path.closepath(),)
|
|
|
def make_dump(s, shift=0, y=0, col=None, bkcol=None, larg=16):
|
c = pyx.canvas.canvas()
|
tlist = []
|
while s:
|
dmp,s = s[:larg-shift],s[larg-shift:]
|
txt = make_dump_txt(shift, y, dmp)
|
tlist.append(txt)
|
shift += len(dmp)
|
if shift >= 16:
|
shift = 0
|
y += 1
|
if col is None:
|
col = pyx.color.rgb.red
|
if bkcol is None:
|
col = pyx.color.rgb.white
|
c.stroke(make_frame(tlist),[col,pyx.deco.filled([bkcol]),pyx.style.linewidth.Thick])
|
for txt in tlist:
|
c.insert(txt)
|
return c, tlist[-1].bbox(), shift, y
|
|
|
last_shift,last_y=0,0.0
|
while t:
|
bkcol = next(backcolor)
|
proto,fields = t.pop()
|
y += 0.5
|
pt = pyx.text.text(XSTART, (YTXT-y)*YMUL, r"\font\cmssfont=cmss10\cmssfont{%s}" % proto.name, [ pyx.text.size.Large])
|
y += 1
|
ptbb=pt.bbox()
|
ptbb.enlarge(pyx.unit.u_pt*2)
|
canvas.stroke(ptbb.path(),[pyx.color.rgb.black, pyx.deco.filled([bkcol])])
|
canvas.insert(pt)
|
for fname, fval, fdump in fields:
|
col = next(forecolor)
|
ft = pyx.text.text(XSTART, (YTXT-y)*YMUL, r"\font\cmssfont=cmss10\cmssfont{%s}" % tex_escape(fname.name))
|
if isinstance(fval, str):
|
if len(fval) > 18:
|
fval = fval[:18]+"[...]"
|
else:
|
fval=""
|
vt = pyx.text.text(XSTART+3, (YTXT-y)*YMUL, r"\font\cmssfont=cmss10\cmssfont{%s}" % tex_escape(fval))
|
y += 1.0
|
if fdump:
|
dt,target,last_shift,last_y = make_dump(fdump, last_shift, last_y, col, bkcol)
|
|
dtb = dt.bbox()
|
dtb=target
|
vtb = vt.bbox()
|
bxvt = make_box(vtb)
|
bxdt = make_box(dtb)
|
dtb.enlarge(pyx.unit.u_pt)
|
try:
|
if yd < 0:
|
cnx = pyx.connector.curve(bxvt,bxdt,absangle1=0, absangle2=-90)
|
else:
|
cnx = pyx.connector.curve(bxvt,bxdt,absangle1=0, absangle2=90)
|
except:
|
pass
|
else:
|
canvas.stroke(cnx,[pyx.style.linewidth.thin,pyx.deco.earrow.small,col])
|
|
canvas.insert(dt)
|
|
canvas.insert(ft)
|
canvas.insert(vt)
|
last_y += layer_shift
|
|
return canvas
|
|
|
|
def extract_padding(self, s):
|
"""
|
DEV: to be overloaded to extract current layer's padding.
|
|
:param str s: the current layer
|
:return: a couple of strings (actual layer, padding)
|
"""
|
return s,None
|
|
def post_dissect(self, s):
|
"""DEV: is called right after the current layer has been dissected"""
|
return s
|
|
def pre_dissect(self, s):
|
"""DEV: is called right before the current layer is dissected"""
|
return s
|
|
def do_dissect(self, s):
|
s = raw(s)
|
_raw = s
|
self.raw_packet_cache_fields = {}
|
for f in self.fields_desc:
|
if not s:
|
break
|
s, fval = f.getfield(self, s)
|
# We need to track fields with mutable values to discard
|
# .raw_packet_cache when needed.
|
if f.islist or f.holds_packets or f.ismutable:
|
self.raw_packet_cache_fields[f.name] = f.do_copy(fval)
|
self.fields[f.name] = fval
|
assert(_raw.endswith(raw(s)))
|
self.raw_packet_cache = _raw[:-len(s)] if s else _raw
|
self.explicit = 1
|
return s
|
|
def do_dissect_payload(self, s):
|
"""
|
Perform the dissection of the layer's payload
|
|
:param str s: the raw layer
|
"""
|
if s:
|
cls = self.guess_payload_class(s)
|
try:
|
p = cls(s, _internal=1, _underlayer=self)
|
except KeyboardInterrupt:
|
raise
|
except:
|
if conf.debug_dissector:
|
if isinstance(cls,type) and issubclass(cls,Packet):
|
log_runtime.error("%s dissector failed" % cls.__name__)
|
else:
|
log_runtime.error("%s.guess_payload_class() returned [%s]" % (self.__class__.__name__,repr(cls)))
|
if cls is not None:
|
raise
|
p = conf.raw_layer(s, _internal=1, _underlayer=self)
|
self.add_payload(p)
|
|
def dissect(self, s):
|
s = self.pre_dissect(s)
|
|
s = self.do_dissect(s)
|
|
s = self.post_dissect(s)
|
|
payl,pad = self.extract_padding(s)
|
self.do_dissect_payload(payl)
|
if pad and conf.padding:
|
self.add_payload(conf.padding_layer(pad))
|
|
|
def guess_payload_class(self, payload):
|
"""
|
DEV: Guesses the next payload class from layer bonds.
|
Can be overloaded to use a different mechanism.
|
|
:param str payload: the layer's payload
|
:return: the payload class
|
"""
|
for t in self.aliastypes:
|
for fval, cls in t.payload_guess:
|
ok = 1
|
for k, v in six.iteritems(fval):
|
if not hasattr(self, k) or v != self.getfieldval(k):
|
ok = 0
|
break
|
if ok:
|
return cls
|
return self.default_payload_class(payload)
|
|
def default_payload_class(self, payload):
|
"""
|
DEV: Returns the default payload class if nothing has been found by the
|
guess_payload_class() method.
|
|
:param str payload: the layer's payload
|
:return: the default payload class define inside the configuration file
|
"""
|
return conf.raw_layer
|
|
def hide_defaults(self):
|
"""Removes fields' values that are the same as default values."""
|
for k, v in list(self.fields.items()): # use list(): self.fields is modified in the loop
|
v = self.fields[k]
|
if k in self.default_fields:
|
if self.default_fields[k] == v:
|
del self.fields[k]
|
self.payload.hide_defaults()
|
|
def clone_with(self, payload=None, **kargs):
|
pkt = self.__class__()
|
pkt.explicit = 1
|
pkt.fields = kargs
|
pkt.default_fields = self.copy_fields_dict(self.default_fields)
|
pkt.overloaded_fields = self.overloaded_fields.copy()
|
pkt.time = self.time
|
pkt.underlayer = self.underlayer
|
pkt.post_transforms = self.post_transforms
|
pkt.raw_packet_cache = self.raw_packet_cache
|
pkt.raw_packet_cache_fields = self.copy_fields_dict(
|
self.raw_packet_cache_fields
|
)
|
if payload is not None:
|
pkt.add_payload(payload)
|
return pkt
|
|
def __iter__(self):
|
def loop(todo, done, self=self):
|
if todo:
|
eltname = todo.pop()
|
elt = self.getfieldval(eltname)
|
if not isinstance(elt, Gen):
|
if self.get_field(eltname).islist:
|
elt = SetGen([elt])
|
else:
|
elt = SetGen(elt)
|
for e in elt:
|
done[eltname]=e
|
for x in loop(todo[:], done):
|
yield x
|
else:
|
if isinstance(self.payload,NoPayload):
|
payloads = [None]
|
else:
|
payloads = self.payload
|
for payl in payloads:
|
done2=done.copy()
|
for k in done2:
|
if isinstance(done2[k], VolatileValue):
|
done2[k] = done2[k]._fix()
|
pkt = self.clone_with(payload=payl, **done2)
|
yield pkt
|
|
if self.explicit or self.raw_packet_cache is not None:
|
todo = []
|
done = self.fields
|
else:
|
todo = [k for (k,v) in itertools.chain(six.iteritems(self.default_fields),
|
six.iteritems(self.overloaded_fields))
|
if isinstance(v, VolatileValue)] + list(self.fields.keys())
|
done = {}
|
return loop(todo, done)
|
|
def __gt__(self, other):
|
"""True if other is an answer from self (self ==> other)."""
|
if isinstance(other, Packet):
|
return other < self
|
elif isinstance(other, bytes):
|
return 1
|
else:
|
raise TypeError((self, other))
|
def __lt__(self, other):
|
"""True if self is an answer from other (other ==> self)."""
|
if isinstance(other, Packet):
|
return self.answers(other)
|
elif isinstance(other, bytes):
|
return 1
|
else:
|
raise TypeError((self, other))
|
|
def __eq__(self, other):
|
if not isinstance(other, self.__class__):
|
return False
|
for f in self.fields_desc:
|
if f not in other.fields_desc:
|
return False
|
if self.getfieldval(f.name) != other.getfieldval(f.name):
|
return False
|
return self.payload == other.payload
|
|
def __ne__(self, other):
|
return not self.__eq__(other)
|
|
def hashret(self):
|
"""DEV: returns a string that has the same value for a request and its answer."""
|
return self.payload.hashret()
|
def answers(self, other):
|
"""DEV: true if self is an answer from other"""
|
if other.__class__ == self.__class__:
|
return self.payload.answers(other.payload)
|
return 0
|
|
def haslayer(self, cls):
|
"""true if self has a layer that is an instance of cls. Superseded by "cls in self" syntax."""
|
if self.__class__ == cls or self.__class__.__name__ == cls:
|
return 1
|
for f in self.packetfields:
|
fvalue_gen = self.getfieldval(f.name)
|
if fvalue_gen is None:
|
continue
|
if not f.islist:
|
fvalue_gen = SetGen(fvalue_gen,_iterpacket=0)
|
for fvalue in fvalue_gen:
|
if isinstance(fvalue, Packet):
|
ret = fvalue.haslayer(cls)
|
if ret:
|
return ret
|
return self.payload.haslayer(cls)
|
|
def getlayer(self, cls, nb=1, _track=None, _subclass=False, **flt):
|
"""Return the nb^th layer that is an instance of cls, matching flt
|
values.
|
|
"""
|
if _subclass:
|
match = lambda cls1, cls2: issubclass(cls1, cls2)
|
else:
|
match = lambda cls1, cls2: cls1 == cls2
|
if isinstance(cls, int):
|
nb = cls+1
|
cls = None
|
if isinstance(cls, str) and "." in cls:
|
ccls,fld = cls.split(".",1)
|
else:
|
ccls,fld = cls,None
|
if cls is None or match(self.__class__, cls) or self.__class__.__name__ == ccls:
|
if all(self.getfieldval(fldname) == fldvalue
|
for fldname, fldvalue in six.iteritems(flt)):
|
if nb == 1:
|
if fld is None:
|
return self
|
else:
|
return self.getfieldval(fld)
|
else:
|
nb -=1
|
for f in self.packetfields:
|
fvalue_gen = self.getfieldval(f.name)
|
if fvalue_gen is None:
|
continue
|
if not f.islist:
|
fvalue_gen = SetGen(fvalue_gen,_iterpacket=0)
|
for fvalue in fvalue_gen:
|
if isinstance(fvalue, Packet):
|
track=[]
|
ret = fvalue.getlayer(cls, nb=nb, _track=track,
|
_subclass=_subclass)
|
if ret is not None:
|
return ret
|
nb = track[0]
|
return self.payload.getlayer(cls, nb=nb, _track=_track,
|
_subclass=_subclass, **flt)
|
|
def firstlayer(self):
|
q = self
|
while q.underlayer is not None:
|
q = q.underlayer
|
return q
|
|
def __getitem__(self, cls):
|
if isinstance(cls, slice):
|
lname = cls.start
|
if cls.stop:
|
ret = self.getlayer(cls.start, nb=cls.stop, **(cls.step or {}))
|
else:
|
ret = self.getlayer(cls.start, **(cls.step or {}))
|
else:
|
lname = cls
|
ret = self.getlayer(cls)
|
if ret is None:
|
if isinstance(lname, Packet_metaclass):
|
lname = lname.__name__
|
elif not isinstance(lname, bytes):
|
lname = repr(lname)
|
raise IndexError("Layer [%s] not found" % lname)
|
return ret
|
|
def __delitem__(self, cls):
|
del(self[cls].underlayer.payload)
|
|
def __setitem__(self, cls, val):
|
self[cls].underlayer.payload = val
|
|
def __contains__(self, cls):
|
""""cls in self" returns true if self has a layer which is an instance of cls."""
|
return self.haslayer(cls)
|
|
def route(self):
|
return (None,None,None)
|
|
def fragment(self, *args, **kargs):
|
return self.payload.fragment(*args, **kargs)
|
|
|
def display(self,*args,**kargs): # Deprecated. Use show()
|
"""Deprecated. Use show() method."""
|
self.show(*args,**kargs)
|
|
def _show_or_dump(self, dump=False, indent=3, lvl="", label_lvl="", first_call=True):
|
"""
|
Internal method that shows or dumps a hierarchical view of a packet.
|
Called by show.
|
|
:param dump: determine if it prints or returns the string value
|
:param int indent: the size of indentation for each layer
|
:param str lvl: additional information about the layer lvl
|
:param str label_lvl: additional information about the layer fields
|
:param first_call: determine if the current function is the first
|
:return: return a hierarchical view if dump, else print it
|
"""
|
|
if dump:
|
from scapy.themes import AnsiColorTheme
|
ct = AnsiColorTheme() # No color for dump output
|
else:
|
ct = conf.color_theme
|
s = "%s%s %s %s \n" % (label_lvl,
|
ct.punct("###["),
|
ct.layer_name(self.name),
|
ct.punct("]###"))
|
for f in self.fields_desc:
|
if isinstance(f, ConditionalField) and not f._evalcond(self):
|
continue
|
if isinstance(f, Emph) or f in conf.emph:
|
ncol = ct.emph_field_name
|
vcol = ct.emph_field_value
|
else:
|
ncol = ct.field_name
|
vcol = ct.field_value
|
fvalue = self.getfieldval(f.name)
|
if isinstance(fvalue, Packet) or (f.islist and f.holds_packets and isinstance(fvalue, list)):
|
s += "%s \\%-10s\\\n" % (label_lvl+lvl, ncol(f.name))
|
fvalue_gen = SetGen(fvalue,_iterpacket=0)
|
for fvalue in fvalue_gen:
|
s += fvalue._show_or_dump(dump=dump, indent=indent, label_lvl=label_lvl+lvl+" |", first_call=False)
|
else:
|
begn = "%s %-10s%s " % (label_lvl+lvl,
|
ncol(f.name),
|
ct.punct("="),)
|
reprval = f.i2repr(self,fvalue)
|
if isinstance(reprval, str):
|
reprval = reprval.replace("\n", "\n"+" "*(len(label_lvl)
|
+len(lvl)
|
+len(f.name)
|
+4))
|
s += "%s%s\n" % (begn,vcol(reprval))
|
if self.payload:
|
s += self.payload._show_or_dump(dump=dump, indent=indent, lvl=lvl+(" "*indent*self.show_indent), label_lvl=label_lvl, first_call=False)
|
|
if first_call and not dump:
|
print(s)
|
else:
|
return s
|
|
def show(self, dump=False, indent=3, lvl="", label_lvl=""):
|
"""
|
Prints or returns (when "dump" is true) a hierarchical view of the
|
packet.
|
|
:param dump: determine if it prints or returns the string value
|
:param int indent: the size of indentation for each layer
|
:param str lvl: additional information about the layer lvl
|
:param str label_lvl: additional information about the layer fields
|
:return: return a hierarchical view if dump, else print it
|
"""
|
return self._show_or_dump(dump, indent, lvl, label_lvl)
|
|
def show2(self, dump=False, indent=3, lvl="", label_lvl=""):
|
"""
|
Prints or returns (when "dump" is true) a hierarchical view of an
|
assembled version of the packet, so that automatic fields are
|
calculated (checksums, etc.)
|
|
:param dump: determine if it prints or returns the string value
|
:param int indent: the size of indentation for each layer
|
:param str lvl: additional information about the layer lvl
|
:param str label_lvl: additional information about the layer fields
|
:return: return a hierarchical view if dump, else print it
|
"""
|
return self.__class__(raw(self)).show(dump, indent, lvl, label_lvl)
|
|
def sprintf(self, fmt, relax=1):
|
"""sprintf(format, [relax=1]) -> str
|
where format is a string that can include directives. A directive begins and
|
ends by % and has the following format %[fmt[r],][cls[:nb].]field%.
|
|
fmt is a classic printf directive, "r" can be appended for raw substitution
|
(ex: IP.flags=0x18 instead of SA), nb is the number of the layer we want
|
(ex: for IP/IP packets, IP:2.src is the src of the upper IP layer).
|
Special case : "%.time%" is the creation time.
|
Ex : p.sprintf("%.time% %-15s,IP.src% -> %-15s,IP.dst% %IP.chksum% "
|
"%03xr,IP.proto% %r,TCP.flags%")
|
|
Moreover, the format string can include conditional statements. A conditional
|
statement looks like : {layer:string} where layer is a layer name, and string
|
is the string to insert in place of the condition if it is true, i.e. if layer
|
is present. If layer is preceded by a "!", the result is inverted. Conditions
|
can be imbricated. A valid statement can be :
|
p.sprintf("This is a{TCP: TCP}{UDP: UDP}{ICMP:n ICMP} packet")
|
p.sprintf("{IP:%IP.dst% {ICMP:%ICMP.type%}{TCP:%TCP.dport%}}")
|
|
A side effect is that, to obtain "{" and "}" characters, you must use
|
"%(" and "%)".
|
"""
|
|
escape = { "%": "%",
|
"(": "{",
|
")": "}" }
|
|
|
# Evaluate conditions
|
while "{" in fmt:
|
i = fmt.rindex("{")
|
j = fmt[i+1:].index("}")
|
cond = fmt[i+1:i+j+1]
|
k = cond.find(":")
|
if k < 0:
|
raise Scapy_Exception("Bad condition in format string: [%s] (read sprintf doc!)"%cond)
|
cond,format = cond[:k],cond[k+1:]
|
res = False
|
if cond[0] == "!":
|
res = True
|
cond = cond[1:]
|
if self.haslayer(cond):
|
res = not res
|
if not res:
|
format = ""
|
fmt = fmt[:i]+format+fmt[i+j+2:]
|
|
# Evaluate directives
|
s = ""
|
while "%" in fmt:
|
i = fmt.index("%")
|
s += fmt[:i]
|
fmt = fmt[i+1:]
|
if fmt and fmt[0] in escape:
|
s += escape[fmt[0]]
|
fmt = fmt[1:]
|
continue
|
try:
|
i = fmt.index("%")
|
sfclsfld = fmt[:i]
|
fclsfld = sfclsfld.split(",")
|
if len(fclsfld) == 1:
|
f = "s"
|
clsfld = fclsfld[0]
|
elif len(fclsfld) == 2:
|
f,clsfld = fclsfld
|
else:
|
raise Scapy_Exception
|
if "." in clsfld:
|
cls,fld = clsfld.split(".")
|
else:
|
cls = self.__class__.__name__
|
fld = clsfld
|
num = 1
|
if ":" in cls:
|
cls,num = cls.split(":")
|
num = int(num)
|
fmt = fmt[i+1:]
|
except:
|
raise Scapy_Exception("Bad format string [%%%s%s]" % (fmt[:25], fmt[25:] and "..."))
|
else:
|
if fld == "time":
|
val = time.strftime("%H:%M:%S.%%06i", time.localtime(self.time)) % int((self.time-int(self.time))*1000000)
|
elif cls == self.__class__.__name__ and hasattr(self, fld):
|
if num > 1:
|
val = self.payload.sprintf("%%%s,%s:%s.%s%%" % (f,cls,num-1,fld), relax)
|
f = "s"
|
elif f[-1] == "r": # Raw field value
|
val = getattr(self,fld)
|
f = f[:-1]
|
if not f:
|
f = "s"
|
else:
|
val = getattr(self,fld)
|
if fld in self.fieldtype:
|
val = self.fieldtype[fld].i2repr(self,val)
|
else:
|
val = self.payload.sprintf("%%%s%%" % sfclsfld, relax)
|
f = "s"
|
s += ("%"+f) % val
|
|
s += fmt
|
return s
|
|
def mysummary(self):
|
"""DEV: can be overloaded to return a string that summarizes the layer.
|
Only one mysummary() is used in a whole packet summary: the one of the upper layer,
|
except if a mysummary() also returns (as a couple) a list of layers whose
|
mysummary() must be called if they are present."""
|
return ""
|
|
def _do_summary(self):
|
found, s, needed = self.payload._do_summary()
|
ret = ""
|
if not found or self.__class__ in needed:
|
ret = self.mysummary()
|
if isinstance(ret, tuple):
|
ret,n = ret
|
needed += n
|
if ret or needed:
|
found = 1
|
if not ret:
|
ret = self.__class__.__name__ if self.show_summary else ""
|
if self.__class__ in conf.emph:
|
impf = []
|
for f in self.fields_desc:
|
if f in conf.emph:
|
impf.append("%s=%s" % (f.name, f.i2repr(self, self.getfieldval(f.name))))
|
ret = "%s [%s]" % (ret," ".join(impf))
|
if ret and s:
|
ret = "%s / %s" % (ret, s)
|
else:
|
ret = "%s%s" % (ret,s)
|
return found,ret,needed
|
|
def summary(self, intern=0):
|
"""Prints a one line summary of a packet."""
|
found,s,needed = self._do_summary()
|
return s
|
|
|
def lastlayer(self,layer=None):
|
"""Returns the uppest layer of the packet"""
|
return self.payload.lastlayer(self)
|
|
def decode_payload_as(self,cls):
|
"""Reassembles the payload and decode it using another packet class"""
|
s = raw(self.payload)
|
self.payload = cls(s, _internal=1, _underlayer=self)
|
pp = self
|
while pp.underlayer is not None:
|
pp = pp.underlayer
|
self.payload.dissection_done(pp)
|
|
def command(self):
|
"""Returns a string representing the command you have to type to obtain the same packet"""
|
f = []
|
for fn,fv in self.fields.items():
|
fld = self.get_field(fn)
|
if isinstance(fv, Packet):
|
fv = fv.command()
|
elif fld.islist and fld.holds_packets and isinstance(fv, list):
|
fv = "[%s]" % ",".join( map(Packet.command, fv))
|
elif isinstance(fld, FlagsField):
|
fv = int(fv)
|
else:
|
fv = repr(fv)
|
f.append("%s=%s" % (fn, fv))
|
c = "%s(%s)" % (self.__class__.__name__, ", ".join(f))
|
pc = self.payload.command()
|
if pc:
|
c += "/"+pc
|
return c
|
|
class NoPayload(Packet):
|
def __new__(cls, *args, **kargs):
|
singl = cls.__dict__.get("__singl__")
|
if singl is None:
|
cls.__singl__ = singl = Packet.__new__(cls)
|
Packet.__init__(singl)
|
return singl
|
def __init__(self, *args, **kargs):
|
pass
|
def dissection_done(self,pkt):
|
return
|
def add_payload(self, payload):
|
raise Scapy_Exception("Can't add payload to NoPayload instance")
|
def remove_payload(self):
|
pass
|
def add_underlayer(self,underlayer):
|
pass
|
def remove_underlayer(self,other):
|
pass
|
def copy(self):
|
return self
|
def __repr__(self):
|
return ""
|
def __str__(self):
|
return ""
|
def __bytes__(self):
|
return b""
|
def __nonzero__(self):
|
return False
|
__bool__ = __nonzero__
|
def do_build(self):
|
return b""
|
def build(self):
|
return b""
|
def build_padding(self):
|
return b""
|
def build_done(self, p):
|
return p
|
def build_ps(self, internal=0):
|
return b"",[]
|
def getfieldval(self, attr):
|
raise AttributeError(attr)
|
def getfield_and_val(self, attr):
|
raise AttributeError(attr)
|
def setfieldval(self, attr, val):
|
raise AttributeError(attr)
|
def delfieldval(self, attr):
|
raise AttributeError(attr)
|
def hide_defaults(self):
|
pass
|
def __iter__(self):
|
return iter([])
|
def __eq__(self, other):
|
if isinstance(other, NoPayload):
|
return True
|
return False
|
def hashret(self):
|
return b""
|
def answers(self, other):
|
return isinstance(other, NoPayload) or isinstance(other, conf.padding_layer)
|
def haslayer(self, cls):
|
return 0
|
def getlayer(self, cls, nb=1, _track=None, **flt):
|
if _track is not None:
|
_track.append(nb)
|
return None
|
def fragment(self, *args, **kargs):
|
raise Scapy_Exception("cannot fragment this packet")
|
def show(self, indent=3, lvl="", label_lvl=""):
|
pass
|
def sprintf(self, fmt, relax):
|
if relax:
|
return "??"
|
else:
|
raise Scapy_Exception("Format not found [%s]"%fmt)
|
def _do_summary(self):
|
return 0,"",[]
|
def lastlayer(self,layer):
|
return layer
|
def command(self):
|
return ""
|
|
####################
|
## packet classes ##
|
####################
|
|
|
class Raw(Packet):
|
name = "Raw"
|
fields_desc = [ StrField("load", "") ]
|
def answers(self, other):
|
return 1
|
# s = raw(other)
|
# t = self.load
|
# l = min(len(s), len(t))
|
# return s[:l] == t[:l]
|
def mysummary(self):
|
cs = conf.raw_summary
|
if cs:
|
if callable(cs):
|
return "Raw %s" % cs(self.load)
|
else:
|
return "Raw %r" % self.load
|
return Packet.mysummary(self)
|
|
class Padding(Raw):
|
name = "Padding"
|
def self_build(self):
|
return b""
|
def build_padding(self):
|
return (raw(self.load) if self.raw_packet_cache is None
|
else self.raw_packet_cache) + self.payload.build_padding()
|
|
conf.raw_layer = Raw
|
conf.padding_layer = Padding
|
if conf.default_l2 is None:
|
conf.default_l2 = Raw
|
|
#################
|
## Bind layers ##
|
#################
|
|
|
def bind_bottom_up(lower, upper, __fval=None, **fval):
|
if __fval is not None:
|
fval.update(__fval)
|
lower.payload_guess = lower.payload_guess[:]
|
lower.payload_guess.append((fval, upper))
|
|
|
def bind_top_down(lower, upper, __fval=None, **fval):
|
if __fval is not None:
|
fval.update(__fval)
|
upper._overload_fields = upper._overload_fields.copy()
|
upper._overload_fields[lower] = fval
|
|
@conf.commands.register
|
def bind_layers(lower, upper, __fval=None, **fval):
|
"""Bind 2 layers on some specific fields' values"""
|
if __fval is not None:
|
fval.update(__fval)
|
bind_top_down(lower, upper, **fval)
|
bind_bottom_up(lower, upper, **fval)
|
|
def split_bottom_up(lower, upper, __fval=None, **fval):
|
if __fval is not None:
|
fval.update(__fval)
|
def do_filter(xxx_todo_changeme,upper=upper,fval=fval):
|
(f,u) = xxx_todo_changeme
|
if u != upper:
|
return True
|
for k in fval:
|
if k not in f or f[k] != fval[k]:
|
return True
|
return False
|
lower.payload_guess = [x for x in lower.payload_guess if do_filter(x)]
|
|
def split_top_down(lower, upper, __fval=None, **fval):
|
if __fval is not None:
|
fval.update(__fval)
|
if lower in upper._overload_fields:
|
ofval = upper._overload_fields[lower]
|
for k in fval:
|
if k not in ofval or ofval[k] != fval[k]:
|
return
|
upper._overload_fields = upper._overload_fields.copy()
|
del(upper._overload_fields[lower])
|
|
@conf.commands.register
|
def split_layers(lower, upper, __fval=None, **fval):
|
"""Split 2 layers previously bound"""
|
if __fval is not None:
|
fval.update(__fval)
|
split_bottom_up(lower, upper, **fval)
|
split_top_down(lower, upper, **fval)
|
|
|
@conf.commands.register
|
def ls(obj=None, case_sensitive=False, verbose=False):
|
"""List available layers, or infos on a given layer class or name"""
|
is_string = isinstance(obj, six.string_types)
|
|
if obj is None or is_string:
|
if obj is None:
|
all_layers = sorted(conf.layers, key=lambda x: x.__name__)
|
else:
|
pattern = re.compile(obj, 0 if case_sensitive else re.I)
|
all_layers = sorted((layer for layer in conf.layers
|
if (pattern.search(layer.__name__ or '')
|
or pattern.search(layer.name or ''))),
|
key=lambda x: x.__name__)
|
for layer in all_layers:
|
print("%-10s : %s" % (layer.__name__, layer._name))
|
|
else:
|
is_pkt = isinstance(obj, Packet)
|
if (isinstance(obj, type) and issubclass(obj, Packet)) or is_pkt:
|
for f in obj.fields_desc:
|
cur_fld = f
|
attrs = []
|
long_attrs = []
|
while isinstance(cur_fld, (Emph, ConditionalField)):
|
if isinstance(cur_fld, ConditionalField):
|
attrs.append(cur_fld.__class__.__name__[:4])
|
cur_fld = cur_fld.fld
|
if verbose and isinstance(cur_fld, EnumField) \
|
and hasattr(cur_fld, "i2s"):
|
if len(cur_fld.i2s) < 50:
|
long_attrs.extend(
|
"%s: %d" % (strval, numval)
|
for numval, strval in
|
sorted(six.iteritems(cur_fld.i2s))
|
)
|
elif isinstance(cur_fld, MultiEnumField):
|
fld_depend = cur_fld.depends_on(obj.__class__
|
if is_pkt else obj)
|
attrs.append("Depends on %s" % fld_depend.name)
|
if verbose:
|
cur_i2s = cur_fld.i2s_multi.get(
|
cur_fld.depends_on(obj if is_pkt else obj()), {}
|
)
|
if len(cur_i2s) < 50:
|
long_attrs.extend(
|
"%s: %d" % (strval, numval)
|
for numval, strval in
|
sorted(six.iteritems(cur_i2s))
|
)
|
elif verbose and isinstance(cur_fld, FlagsField):
|
names = cur_fld.names
|
long_attrs.append(", ".join(names))
|
class_name = "%s (%s)" % (
|
cur_fld.__class__.__name__,
|
", ".join(attrs)) if attrs else cur_fld.__class__.__name__
|
if isinstance(cur_fld, BitField):
|
class_name += " (%d bit%s)" % (cur_fld.size,
|
"s" if cur_fld.size > 1
|
else "")
|
print("%-10s : %-35s =" % (f.name, class_name), end=' ')
|
if is_pkt:
|
print("%-15r" % (getattr(obj, f.name),), end=' ')
|
print("(%r)" % (f.default,))
|
for attr in long_attrs:
|
print("%-15s%s" % ("", attr))
|
if is_pkt and not isinstance(obj.payload, NoPayload):
|
print("--")
|
ls(obj.payload)
|
|
else:
|
print("Not a packet class or name. Type 'ls()' to list packet classes.")
|
|
|
|
#############
|
## Fuzzing ##
|
#############
|
|
@conf.commands.register
|
def fuzz(p, _inplace=0):
|
"""Transform a layer into a fuzzy layer by replacing some default values by random objects"""
|
if not _inplace:
|
p = p.copy()
|
q = p
|
while not isinstance(q, NoPayload):
|
for f in q.fields_desc:
|
if isinstance(f, PacketListField):
|
for r in getattr(q, f.name):
|
print("fuzzing", repr(r))
|
fuzz(r, _inplace=1)
|
elif f.default is not None:
|
if not isinstance(f, ConditionalField) or f._evalcond(q):
|
rnd = f.randval()
|
if rnd is not None:
|
q.default_fields[f.name] = rnd
|
q = q.payload
|
return p
|
