705 lines
27 KiB
Python
Executable file
705 lines
27 KiB
Python
Executable file
# This file is part of Scapy
|
|
# See http://www.secdev.org/projects/scapy for more information
|
|
# Copyright (C) Philippe Biondi <phil@secdev.org>
|
|
# This program is published under a GPLv2 license
|
|
|
|
"""
|
|
PacketList: holds several packets and allows to do operations on them.
|
|
"""
|
|
|
|
|
|
from __future__ import absolute_import
|
|
from __future__ import print_function
|
|
import os
|
|
from collections import defaultdict
|
|
|
|
from scapy.compat import lambda_tuple_converter
|
|
from scapy.config import conf
|
|
from scapy.base_classes import BasePacket, BasePacketList, _CanvasDumpExtended
|
|
from scapy.fields import IPField, ShortEnumField, PacketField
|
|
from scapy.utils import do_graph, hexdump, make_table, make_lined_table, \
|
|
make_tex_table, issubtype
|
|
from scapy.extlib import plt, Line2D, \
|
|
MATPLOTLIB_INLINED, MATPLOTLIB_DEFAULT_PLOT_KARGS
|
|
from functools import reduce
|
|
import scapy.modules.six as six
|
|
from scapy.modules.six.moves import range, zip
|
|
from scapy.compat import Optional, List, Union, Tuple, Dict, Any, Callable
|
|
from scapy.packet import Packet
|
|
#############
|
|
# Results #
|
|
#############
|
|
|
|
|
|
class PacketList(BasePacketList, _CanvasDumpExtended):
|
|
__slots__ = ["stats", "res", "listname"]
|
|
|
|
def __init__(self, res=None, name="PacketList", stats=None):
|
|
"""create a packet list from a list of packets
|
|
res: the list of packets
|
|
stats: a list of classes that will appear in the stats (defaults to [TCP,UDP,ICMP])""" # noqa: E501
|
|
if stats is None:
|
|
stats = conf.stats_classic_protocols
|
|
self.stats = stats
|
|
if res is None:
|
|
res = []
|
|
elif isinstance(res, PacketList):
|
|
res = res.res
|
|
self.res = res
|
|
self.listname = name
|
|
|
|
def __len__(self):
|
|
# type: () -> int
|
|
return len(self.res)
|
|
|
|
def _elt2pkt(self, elt):
|
|
# type: (Packet) -> Packet
|
|
return elt
|
|
|
|
def _elt2sum(self, elt):
|
|
# type: (Packet) -> str
|
|
return elt.summary()
|
|
|
|
def _elt2show(self, elt):
|
|
# type: (Packet) -> str
|
|
return self._elt2sum(elt)
|
|
|
|
def __repr__(self):
|
|
# type: () -> str
|
|
stats = {x: 0 for x in self.stats}
|
|
other = 0
|
|
for r in self.res:
|
|
f = 0
|
|
for p in stats:
|
|
if self._elt2pkt(r).haslayer(p):
|
|
stats[p] += 1
|
|
f = 1
|
|
break
|
|
if not f:
|
|
other += 1
|
|
s = ""
|
|
ct = conf.color_theme
|
|
for p in self.stats:
|
|
s += " %s%s%s" % (ct.packetlist_proto(p._name),
|
|
ct.punct(":"),
|
|
ct.packetlist_value(stats[p]))
|
|
s += " %s%s%s" % (ct.packetlist_proto("Other"),
|
|
ct.punct(":"),
|
|
ct.packetlist_value(other))
|
|
return "%s%s%s%s%s" % (ct.punct("<"),
|
|
ct.packetlist_name(self.listname),
|
|
ct.punct(":"),
|
|
s,
|
|
ct.punct(">"))
|
|
|
|
def __getstate__(self):
|
|
# type: () -> Dict[str, Union[List[PacketField], List[Packet], str]]
|
|
"""
|
|
Creates a basic representation of the instance, used in
|
|
conjunction with __setstate__() e.g. by pickle
|
|
|
|
:returns: dict representing this instance
|
|
"""
|
|
state = {
|
|
'res': self.res,
|
|
'stats': self.stats,
|
|
'listname': self.listname
|
|
}
|
|
return state
|
|
|
|
def __setstate__(self, state):
|
|
# type: (Dict[str, Union[List[PacketField], List[Packet], str]]) -> None # noqa: E501
|
|
"""
|
|
Sets instance attributes to values given by state, used in
|
|
conjunction with __getstate__() e.g. by pickle
|
|
|
|
:param state: dict representing this instance
|
|
"""
|
|
self.res = state['res']
|
|
self.stats = state['stats']
|
|
self.listname = state['listname']
|
|
|
|
def __getattr__(self, attr):
|
|
# type: (str) -> Any
|
|
return getattr(self.res, attr)
|
|
|
|
def __getitem__(self, item):
|
|
if issubtype(item, BasePacket):
|
|
return self.__class__([x for x in self.res if item in self._elt2pkt(x)], # noqa: E501
|
|
name="%s from %s" % (item.__name__, self.listname)) # noqa: E501
|
|
if isinstance(item, slice):
|
|
return self.__class__(self.res.__getitem__(item),
|
|
name="mod %s" % self.listname)
|
|
return self.res.__getitem__(item)
|
|
|
|
def __add__(self, other):
|
|
# type: (PacketList) -> PacketList
|
|
return self.__class__(self.res + other.res,
|
|
name="%s+%s" % (self.listname, other.listname))
|
|
|
|
def summary(self, prn=None, lfilter=None):
|
|
# type: (Optional[Callable], Optional[Callable]) -> None
|
|
"""prints a summary of each packet
|
|
|
|
:param prn: function to apply to each packet instead of
|
|
lambda x:x.summary()
|
|
:param lfilter: truth function to apply to each packet to decide
|
|
whether it will be displayed
|
|
"""
|
|
for r in self.res:
|
|
if lfilter is not None:
|
|
if not lfilter(r):
|
|
continue
|
|
if prn is None:
|
|
print(self._elt2sum(r))
|
|
else:
|
|
print(prn(r))
|
|
|
|
def nsummary(self, prn=None, lfilter=None):
|
|
# type: (Optional[Callable], Optional[Callable]) -> None
|
|
"""prints a summary of each packet with the packet's number
|
|
|
|
:param prn: function to apply to each packet instead of
|
|
lambda x:x.summary()
|
|
:param lfilter: truth function to apply to each packet to decide
|
|
whether it will be displayed
|
|
"""
|
|
for i, res in enumerate(self.res):
|
|
if lfilter is not None:
|
|
if not lfilter(res):
|
|
continue
|
|
print(conf.color_theme.id(i, fmt="%04i"), end=' ')
|
|
if prn is None:
|
|
print(self._elt2sum(res))
|
|
else:
|
|
print(prn(res))
|
|
|
|
def display(self): # Deprecated. Use show()
|
|
"""deprecated. is show()"""
|
|
self.show()
|
|
|
|
def show(self, *args, **kargs):
|
|
# type: (Any, Any) -> None
|
|
"""Best way to display the packet list. Defaults to nsummary() method""" # noqa: E501
|
|
return self.nsummary(*args, **kargs)
|
|
|
|
def filter(self, func):
|
|
# type: (Callable) -> PacketList
|
|
"""Returns a packet list filtered by a truth function. This truth
|
|
function has to take a packet as the only argument and return a boolean value.""" # noqa: E501
|
|
return self.__class__([x for x in self.res if func(x)],
|
|
name="filtered %s" % self.listname)
|
|
|
|
def make_table(self, *args, **kargs):
|
|
# type: (Any, Any) -> None
|
|
"""Prints a table using a function that returns for each packet its head column value, head row value and displayed value # noqa: E501
|
|
ex: p.make_table(lambda x:(x[IP].dst, x[TCP].dport, x[TCP].sprintf("%flags%")) """ # noqa: E501
|
|
return make_table(self.res, *args, **kargs)
|
|
|
|
def make_lined_table(self, *args, **kargs):
|
|
# type: (Any, Any) -> None
|
|
"""Same as make_table, but print a table with lines"""
|
|
return make_lined_table(self.res, *args, **kargs)
|
|
|
|
def make_tex_table(self, *args, **kargs):
|
|
# type: (Any, Any) -> None
|
|
"""Same as make_table, but print a table with LaTeX syntax"""
|
|
return make_tex_table(self.res, *args, **kargs)
|
|
|
|
def plot(self, f, lfilter=None, plot_xy=False, **kargs):
|
|
# type: (Callable, Optional[Callable], bool, Any) -> Line2D
|
|
"""Applies a function to each packet to get a value that will be plotted
|
|
with matplotlib. A list of matplotlib.lines.Line2D is returned.
|
|
|
|
lfilter: a truth function that decides whether a packet must be plotted
|
|
"""
|
|
|
|
# Python 2 backward compatibility
|
|
f = lambda_tuple_converter(f)
|
|
lfilter = lambda_tuple_converter(lfilter)
|
|
|
|
# Get the list of packets
|
|
if lfilter is None:
|
|
lst_pkts = [f(*e) for e in self.res]
|
|
else:
|
|
lst_pkts = [f(*e) for e in self.res if lfilter(*e)]
|
|
|
|
# Mimic the default gnuplot output
|
|
if kargs == {}:
|
|
kargs = MATPLOTLIB_DEFAULT_PLOT_KARGS
|
|
if plot_xy:
|
|
lines = plt.plot(*zip(*lst_pkts), **kargs)
|
|
else:
|
|
lines = plt.plot(lst_pkts, **kargs)
|
|
|
|
# Call show() if matplotlib is not inlined
|
|
if not MATPLOTLIB_INLINED:
|
|
plt.show()
|
|
|
|
return lines
|
|
|
|
def diffplot(self, f, delay=1, lfilter=None, **kargs):
|
|
# type: (Callable, int, Optional[Callable], Any) -> Line2D
|
|
"""diffplot(f, delay=1, lfilter=None)
|
|
Applies a function to couples (l[i],l[i+delay])
|
|
|
|
A list of matplotlib.lines.Line2D is returned.
|
|
"""
|
|
|
|
# Get the list of packets
|
|
if lfilter is None:
|
|
lst_pkts = [f(self.res[i], self.res[i + 1])
|
|
for i in range(len(self.res) - delay)]
|
|
else:
|
|
lst_pkts = [f(self.res[i], self.res[i + 1])
|
|
for i in range(len(self.res) - delay)
|
|
if lfilter(self.res[i])]
|
|
|
|
# Mimic the default gnuplot output
|
|
if kargs == {}:
|
|
kargs = MATPLOTLIB_DEFAULT_PLOT_KARGS
|
|
lines = plt.plot(lst_pkts, **kargs)
|
|
|
|
# Call show() if matplotlib is not inlined
|
|
if not MATPLOTLIB_INLINED:
|
|
plt.show()
|
|
|
|
return lines
|
|
|
|
def multiplot(self, f, lfilter=None, plot_xy=False, **kargs):
|
|
# type: (Callable, Optional[Callable], bool, Any) -> Line2D
|
|
"""Uses a function that returns a label and a value for this label, then
|
|
plots all the values label by label.
|
|
|
|
A list of matplotlib.lines.Line2D is returned.
|
|
"""
|
|
|
|
# Python 2 backward compatibility
|
|
f = lambda_tuple_converter(f)
|
|
lfilter = lambda_tuple_converter(lfilter)
|
|
|
|
# Get the list of packets
|
|
if lfilter is None:
|
|
lst_pkts = (f(*e) for e in self.res)
|
|
else:
|
|
lst_pkts = (f(*e) for e in self.res if lfilter(*e))
|
|
|
|
# Apply the function f to the packets
|
|
d = {} # type: Dict[str, List[float]]
|
|
for k, v in lst_pkts:
|
|
d.setdefault(k, []).append(v)
|
|
|
|
# Mimic the default gnuplot output
|
|
if not kargs:
|
|
kargs = MATPLOTLIB_DEFAULT_PLOT_KARGS
|
|
|
|
if plot_xy:
|
|
lines = [plt.plot(*zip(*pl), **dict(kargs, label=k))
|
|
for k, pl in six.iteritems(d)]
|
|
else:
|
|
lines = [plt.plot(pl, **dict(kargs, label=k))
|
|
for k, pl in six.iteritems(d)]
|
|
plt.legend(loc="center right", bbox_to_anchor=(1.5, 0.5))
|
|
|
|
# Call show() if matplotlib is not inlined
|
|
if not MATPLOTLIB_INLINED:
|
|
plt.show()
|
|
|
|
return lines
|
|
|
|
def rawhexdump(self):
|
|
# type: (Optional[Callable]) -> None
|
|
"""Prints an hexadecimal dump of each packet in the list"""
|
|
for p in self:
|
|
hexdump(self._elt2pkt(p))
|
|
|
|
def hexraw(self, lfilter=None):
|
|
# type: (Optional[Callable]) -> None
|
|
"""Same as nsummary(), except that if a packet has a Raw layer, it will be hexdumped # noqa: E501
|
|
lfilter: a truth function that decides whether a packet must be displayed""" # noqa: E501
|
|
for i, res in enumerate(self.res):
|
|
p = self._elt2pkt(res)
|
|
if lfilter is not None and not lfilter(p):
|
|
continue
|
|
print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
|
|
p.sprintf("%.time%"),
|
|
self._elt2sum(res)))
|
|
if p.haslayer(conf.raw_layer):
|
|
hexdump(p.getlayer(conf.raw_layer).load)
|
|
|
|
def hexdump(self, lfilter=None):
|
|
# type: (Optional[Callable]) -> None
|
|
"""Same as nsummary(), except that packets are also hexdumped
|
|
lfilter: a truth function that decides whether a packet must be displayed""" # noqa: E501
|
|
for i, res in enumerate(self.res):
|
|
p = self._elt2pkt(res)
|
|
if lfilter is not None and not lfilter(p):
|
|
continue
|
|
print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
|
|
p.sprintf("%.time%"),
|
|
self._elt2sum(res)))
|
|
hexdump(p)
|
|
|
|
def padding(self, lfilter=None):
|
|
# type: (Optional[Callable]) -> None
|
|
"""Same as hexraw(), for Padding layer"""
|
|
for i, res in enumerate(self.res):
|
|
p = self._elt2pkt(res)
|
|
if p.haslayer(conf.padding_layer):
|
|
if lfilter is None or lfilter(p):
|
|
print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
|
|
p.sprintf("%.time%"),
|
|
self._elt2sum(res)))
|
|
hexdump(p.getlayer(conf.padding_layer).load)
|
|
|
|
def nzpadding(self, lfilter=None):
|
|
# type: (Optional[Callable]) -> None
|
|
"""Same as padding() but only non null padding"""
|
|
for i, res in enumerate(self.res):
|
|
p = self._elt2pkt(res)
|
|
if p.haslayer(conf.padding_layer):
|
|
pad = p.getlayer(conf.padding_layer).load
|
|
if pad == pad[0] * len(pad):
|
|
continue
|
|
if lfilter is None or lfilter(p):
|
|
print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
|
|
p.sprintf("%.time%"),
|
|
self._elt2sum(res)))
|
|
hexdump(p.getlayer(conf.padding_layer).load)
|
|
|
|
def conversations(self, getsrcdst=None, **kargs):
|
|
"""Graphes a conversations between sources and destinations and display it
|
|
(using graphviz and imagemagick)
|
|
|
|
:param getsrcdst: a function that takes an element of the list and
|
|
returns the source, the destination and optionally
|
|
a label. By default, returns the IP source and
|
|
destination from IP and ARP layers
|
|
:param type: output type (svg, ps, gif, jpg, etc.), passed to dot's
|
|
"-T" option
|
|
:param target: filename or redirect. Defaults pipe to Imagemagick's
|
|
display program
|
|
:param prog: which graphviz program to use
|
|
"""
|
|
if getsrcdst is None:
|
|
def getsrcdst(pkt):
|
|
"""Extract src and dst addresses"""
|
|
if 'IP' in pkt:
|
|
return (pkt['IP'].src, pkt['IP'].dst)
|
|
if 'IPv6' in pkt:
|
|
return (pkt['IPv6'].src, pkt['IPv6'].dst)
|
|
if 'ARP' in pkt:
|
|
return (pkt['ARP'].psrc, pkt['ARP'].pdst)
|
|
raise TypeError()
|
|
conv = {}
|
|
for p in self.res:
|
|
p = self._elt2pkt(p)
|
|
try:
|
|
c = getsrcdst(p)
|
|
except Exception:
|
|
# No warning here: it's OK that getsrcdst() raises an
|
|
# exception, since it might be, for example, a
|
|
# function that expects a specific layer in each
|
|
# packet. The try/except approach is faster and
|
|
# considered more Pythonic than adding tests.
|
|
continue
|
|
if len(c) == 3:
|
|
conv.setdefault(c[:2], set()).add(c[2])
|
|
else:
|
|
conv[c] = conv.get(c, 0) + 1
|
|
gr = 'digraph "conv" {\n'
|
|
for (s, d), l in six.iteritems(conv):
|
|
gr += '\t "%s" -> "%s" [label="%s"]\n' % (
|
|
s, d, ', '.join(str(x) for x in l) if isinstance(l, set) else l
|
|
)
|
|
gr += "}\n"
|
|
return do_graph(gr, **kargs)
|
|
|
|
def afterglow(self, src=None, event=None, dst=None, **kargs):
|
|
# type: (Optional[Callable], Optional[Callable], Optional[Callable], Any) -> None # noqa: E501
|
|
"""Experimental clone attempt of http://sourceforge.net/projects/afterglow
|
|
each datum is reduced as src -> event -> dst and the data are graphed.
|
|
by default we have IP.src -> IP.dport -> IP.dst"""
|
|
if src is None:
|
|
src = lambda x: x['IP'].src
|
|
if event is None:
|
|
event = lambda x: x['IP'].dport
|
|
if dst is None:
|
|
dst = lambda x: x['IP'].dst
|
|
sl = {} # type: Dict[IPField, Tuple[int, List[ShortEnumField]]]
|
|
el = {} # type: Dict[ShortEnumField, Tuple[int, List[IPField]]]
|
|
dl = {} # type: Dict[IPField, ShortEnumField]
|
|
for i in self.res:
|
|
try:
|
|
s, e, d = src(i), event(i), dst(i)
|
|
if s in sl:
|
|
n, lst = sl[s]
|
|
n += 1
|
|
if e not in lst:
|
|
lst.append(e)
|
|
sl[s] = (n, lst)
|
|
else:
|
|
sl[s] = (1, [e])
|
|
if e in el:
|
|
n, lst = el[e]
|
|
n += 1
|
|
if d not in lst:
|
|
lst.append(d)
|
|
el[e] = (n, lst)
|
|
else:
|
|
el[e] = (1, [d])
|
|
dl[d] = dl.get(d, 0) + 1
|
|
except Exception:
|
|
continue
|
|
|
|
def minmax(x):
|
|
m, M = reduce(lambda a, b: (min(a[0], b[0]), max(a[1], b[1])),
|
|
((a, a) for a in x))
|
|
if m == M:
|
|
m = 0
|
|
if M == 0:
|
|
M = 1
|
|
return m, M
|
|
|
|
mins, maxs = minmax(x for x, _ in six.itervalues(sl))
|
|
mine, maxe = minmax(x for x, _ in six.itervalues(el))
|
|
mind, maxd = minmax(six.itervalues(dl))
|
|
|
|
gr = 'digraph "afterglow" {\n\tedge [len=2.5];\n'
|
|
|
|
gr += "# src nodes\n"
|
|
for s in sl:
|
|
n, _ = sl[s]
|
|
n = 1 + float(n - mins) / (maxs - mins)
|
|
gr += '"src.%s" [label = "%s", shape=box, fillcolor="#FF0000", style=filled, fixedsize=1, height=%.2f,width=%.2f];\n' % (repr(s), repr(s), n, n) # noqa: E501
|
|
gr += "# event nodes\n"
|
|
for e in el:
|
|
n, _ = el[e]
|
|
n = 1 + float(n - mine) / (maxe - mine)
|
|
gr += '"evt.%s" [label = "%s", shape=circle, fillcolor="#00FFFF", style=filled, fixedsize=1, height=%.2f, width=%.2f];\n' % (repr(e), repr(e), n, n) # noqa: E501
|
|
for d in dl:
|
|
n = dl[d]
|
|
n = 1 + float(n - mind) / (maxd - mind)
|
|
gr += '"dst.%s" [label = "%s", shape=triangle, fillcolor="#0000ff", style=filled, fixedsize=1, height=%.2f, width=%.2f];\n' % (repr(d), repr(d), n, n) # noqa: E501
|
|
|
|
gr += "###\n"
|
|
for s in sl:
|
|
n, lst = sl[s]
|
|
for e in lst:
|
|
gr += ' "src.%s" -> "evt.%s";\n' % (repr(s), repr(e))
|
|
for e in el:
|
|
n, lst = el[e]
|
|
for d in lst:
|
|
gr += ' "evt.%s" -> "dst.%s";\n' % (repr(e), repr(d))
|
|
|
|
gr += "}"
|
|
return do_graph(gr, **kargs)
|
|
|
|
def canvas_dump(self, **kargs):
|
|
# type: (Any) -> Any # Using Any since pyx is imported later
|
|
import pyx
|
|
d = pyx.document.document()
|
|
len_res = len(self.res)
|
|
for i, res in enumerate(self.res):
|
|
c = self._elt2pkt(res).canvas_dump(**kargs)
|
|
cbb = c.bbox()
|
|
c.text(cbb.left(), cbb.top() + 1, r"\font\cmssfont=cmss12\cmssfont{Frame %i/%i}" % (i, len_res), [pyx.text.size.LARGE]) # noqa: E501
|
|
if conf.verb >= 2:
|
|
os.write(1, b".")
|
|
d.append(pyx.document.page(c, paperformat=pyx.document.paperformat.A4, # noqa: E501
|
|
margin=1 * pyx.unit.t_cm,
|
|
fittosize=1))
|
|
return d
|
|
|
|
def sr(self, multi=0):
|
|
# type: (int) -> Tuple[SndRcvList, PacketList]
|
|
"""sr([multi=1]) -> (SndRcvList, PacketList)
|
|
Matches packets in the list and return ( (matched couples), (unmatched packets) )""" # noqa: E501
|
|
remain = self.res[:]
|
|
sr = []
|
|
i = 0
|
|
while i < len(remain):
|
|
s = remain[i]
|
|
j = i
|
|
while j < len(remain) - 1:
|
|
j += 1
|
|
r = remain[j]
|
|
if r.answers(s):
|
|
sr.append((s, r))
|
|
if multi:
|
|
remain[i]._answered = 1
|
|
remain[j]._answered = 2
|
|
continue
|
|
del(remain[j])
|
|
del(remain[i])
|
|
i -= 1
|
|
break
|
|
i += 1
|
|
if multi:
|
|
remain = [x for x in remain if not hasattr(x, "_answered")]
|
|
return SndRcvList(sr), PacketList(remain)
|
|
|
|
def sessions(self, session_extractor=None):
|
|
if session_extractor is None:
|
|
def session_extractor(p):
|
|
"""Extract sessions from packets"""
|
|
if 'Ether' in p:
|
|
if 'IP' in p or 'IPv6' in p:
|
|
ip_src_fmt = "{IP:%IP.src%}{IPv6:%IPv6.src%}"
|
|
ip_dst_fmt = "{IP:%IP.dst%}{IPv6:%IPv6.dst%}"
|
|
addr_fmt = (ip_src_fmt, ip_dst_fmt)
|
|
if 'TCP' in p:
|
|
fmt = "TCP {}:%r,TCP.sport% > {}:%r,TCP.dport%"
|
|
elif 'UDP' in p:
|
|
fmt = "UDP {}:%r,UDP.sport% > {}:%r,UDP.dport%"
|
|
elif 'ICMP' in p:
|
|
fmt = "ICMP {} > {} type=%r,ICMP.type% code=%r," \
|
|
"ICMP.code% id=%ICMP.id%"
|
|
elif 'ICMPv6' in p:
|
|
fmt = "ICMPv6 {} > {} type=%r,ICMPv6.type% " \
|
|
"code=%r,ICMPv6.code%"
|
|
elif 'IPv6' in p:
|
|
fmt = "IPv6 {} > {} nh=%IPv6.nh%"
|
|
else:
|
|
fmt = "IP {} > {} proto=%IP.proto%"
|
|
return p.sprintf(fmt.format(*addr_fmt))
|
|
elif 'ARP' in p:
|
|
return p.sprintf("ARP %ARP.psrc% > %ARP.pdst%")
|
|
else:
|
|
return p.sprintf("Ethernet type=%04xr,Ether.type%")
|
|
return "Other"
|
|
sessions = defaultdict(self.__class__)
|
|
for p in self.res:
|
|
sess = session_extractor(self._elt2pkt(p))
|
|
sessions[sess].append(p)
|
|
return dict(sessions)
|
|
|
|
def replace(self, *args, **kargs):
|
|
# type: (Any, Any) -> PacketList
|
|
"""
|
|
lst.replace(<field>,[<oldvalue>,]<newvalue>)
|
|
lst.replace( (fld,[ov],nv),(fld,[ov,]nv),...)
|
|
if ov is None, all values are replaced
|
|
ex:
|
|
lst.replace( IP.src, "192.168.1.1", "10.0.0.1" )
|
|
lst.replace( IP.ttl, 64 )
|
|
lst.replace( (IP.ttl, 64), (TCP.sport, 666, 777), )
|
|
"""
|
|
delete_checksums = kargs.get("delete_checksums", False)
|
|
x = PacketList(name="Replaced %s" % self.listname)
|
|
if not isinstance(args[0], tuple):
|
|
args = (args,)
|
|
for p in self.res:
|
|
p = self._elt2pkt(p)
|
|
copied = False
|
|
for scheme in args:
|
|
fld = scheme[0]
|
|
old = scheme[1] # not used if len(scheme) == 2
|
|
new = scheme[-1]
|
|
for o in fld.owners:
|
|
if o in p:
|
|
if len(scheme) == 2 or p[o].getfieldval(fld.name) == old: # noqa: E501
|
|
if not copied:
|
|
p = p.copy()
|
|
if delete_checksums:
|
|
p.delete_checksums()
|
|
copied = True
|
|
setattr(p[o], fld.name, new)
|
|
x.append(p)
|
|
return x
|
|
|
|
def getlayer(self, cls, # type: Packet
|
|
nb=None, # type: Optional[int]
|
|
flt=None, # type: Optional[Dict[str, Any]]
|
|
name=None, # type: Optional[str]
|
|
stats=None # type: Optional[List[Packet]]
|
|
):
|
|
# type: (...) -> PacketList
|
|
"""Returns the packet list from a given layer.
|
|
|
|
See ``Packet.getlayer`` for more info.
|
|
|
|
:param cls: search for a layer that is an instance of ``cls``
|
|
:type cls: Type[scapy.packet.Packet]
|
|
|
|
:param nb: return the nb^th layer that is an instance of ``cls``
|
|
:type nb: Optional[int]
|
|
|
|
:param flt: filter parameters for ``Packet.getlayer``
|
|
:type flt: Optional[Dict[str, Any]]
|
|
|
|
:param name: optional name for the new PacketList
|
|
:type name: Optional[str]
|
|
|
|
:param stats: optional list of protocols to give stats on; if not
|
|
specified, inherits from this PacketList.
|
|
:type stats: Optional[List[Type[scapy.packet.Packet]]]
|
|
:rtype: scapy.plist.PacketList
|
|
"""
|
|
if name is None:
|
|
name = "{} layer {}".format(self.listname, cls.__name__)
|
|
if stats is None:
|
|
stats = self.stats
|
|
|
|
getlayer_arg = {} # type: Dict[str, Any]
|
|
if flt is not None:
|
|
getlayer_arg.update(flt)
|
|
getlayer_arg['cls'] = cls
|
|
if nb is not None:
|
|
getlayer_arg['nb'] = nb
|
|
|
|
# Only return non-None getlayer results
|
|
return PacketList([
|
|
pc for pc in (p.getlayer(**getlayer_arg) for p in self.res)
|
|
if pc is not None],
|
|
name, stats
|
|
)
|
|
|
|
def convert_to(self, other_cls, name=None, stats=None):
|
|
# type: (Packet, Optional[str], Optional[List[Packet]]) -> PacketList
|
|
"""Converts all packets to another type.
|
|
|
|
See ``Packet.convert_to`` for more info.
|
|
|
|
:param other_cls: reference to a Packet class to convert to
|
|
:type other_cls: Type[scapy.packet.Packet]
|
|
|
|
:param name: optional name for the new PacketList
|
|
:type name: Optional[str]
|
|
|
|
:param stats: optional list of protocols to give stats on;
|
|
if not specified, inherits from this PacketList.
|
|
:type stats: Optional[List[Type[scapy.packet.Packet]]]
|
|
|
|
:rtype: scapy.plist.PacketList
|
|
"""
|
|
if name is None:
|
|
name = "{} converted to {}".format(
|
|
self.listname, other_cls.__name__)
|
|
if stats is None:
|
|
stats = self.stats
|
|
|
|
return PacketList(
|
|
[p.convert_to(other_cls) for p in self.res],
|
|
name, stats
|
|
)
|
|
|
|
|
|
class SndRcvList(PacketList):
|
|
__slots__ = [] # type: List[str]
|
|
|
|
def __init__(self,
|
|
res=None, # type: Optional[Union[List[Packet], PacketList]]
|
|
name="Results", # type: str
|
|
stats=None # type: Optional[List[Packet]]
|
|
):
|
|
# type: (...) -> None
|
|
PacketList.__init__(self, res, name, stats)
|
|
|
|
def _elt2pkt(self, elt):
|
|
# type: (Tuple[Packet, Packet]) -> Packet
|
|
return elt[1]
|
|
|
|
def _elt2sum(self, elt):
|
|
# type: (Tuple[Packet, Packet]) -> str
|
|
return "%s ==> %s" % (elt[0].summary(), elt[1].summary())
|