Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/plist.py: 18%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
4# Copyright (C) Philippe Biondi <phil@secdev.org>
6"""
7PacketList: holds several packets and allows to do operations on them.
8"""
11import os
12from collections import defaultdict
13from typing import Sequence, NamedTuple
15from scapy.config import conf
16from scapy.base_classes import (
17 BasePacket,
18 BasePacketList,
19 PacketList_metaclass,
20 SetGen,
21 _CanvasDumpExtended,
22)
23from scapy.utils import do_graph, graphviz_escape, hexdump, make_table, \
24 make_lined_table, make_tex_table, issubtype
25from functools import reduce
27# typings
28from typing import (
29 Any,
30 Callable,
31 DefaultDict,
32 Dict,
33 Generic,
34 Iterator,
35 List,
36 Optional,
37 Tuple,
38 Type,
39 TypeVar,
40 Union,
41 TYPE_CHECKING,
42)
43from scapy.packet import Packet
45try:
46 import pyx
47except ImportError:
48 pass
50if TYPE_CHECKING:
51 from scapy.libs.matplot import Line2D
53#############
54# Results #
55#############
58QueryAnswer = NamedTuple(
59 "QueryAnswer",
60 [("query", Packet), ("answer", Packet)]
61)
63_Inner = TypeVar("_Inner", Packet, QueryAnswer)
66class _PacketList(Generic[_Inner], metaclass=PacketList_metaclass):
67 __slots__ = ["stats", "res", "listname"]
69 def __init__(self,
70 res=None, # type: Optional[Union[_PacketList[_Inner], List[_Inner]]] # noqa: E501
71 name="PacketList", # type: str
72 stats=None # type: Optional[List[Type[Packet]]]
73 ):
74 # type: (...) -> None
75 """create a packet list from a list of packets
76 res: the list of packets
77 stats: a list of classes that will appear in the stats (defaults to [TCP,UDP,ICMP])""" # noqa: E501
78 if stats is None:
79 stats = conf.stats_classic_protocols
80 self.stats = stats
81 if res is None:
82 self.res = [] # type: List[_Inner]
83 elif isinstance(res, _PacketList):
84 self.res = res.res
85 else:
86 self.res = res
87 self.listname = name
89 def __len__(self):
90 # type: () -> int
91 return len(self.res)
93 def _elt2pkt(self, elt):
94 # type: (_Inner) -> Packet
95 return elt # type: ignore
97 def _elt2sum(self, elt):
98 # type: (_Inner) -> str
99 return elt.summary() # type: ignore
101 def _elt2show(self, elt):
102 # type: (_Inner) -> str
103 return self._elt2sum(elt)
105 def __repr__(self):
106 # type: () -> str
107 stats = {x: 0 for x in self.stats}
108 other = 0
109 for r in self.res:
110 f = 0
111 for p in stats:
112 if self._elt2pkt(r).haslayer(p):
113 stats[p] += 1
114 f = 1
115 break
116 if not f:
117 other += 1
118 s = ""
119 ct = conf.color_theme
120 for p in self.stats:
121 s += " %s%s%s" % (ct.packetlist_proto(p._name),
122 ct.punct(":"),
123 ct.packetlist_value(stats[p]))
124 s += " %s%s%s" % (ct.packetlist_proto("Other"),
125 ct.punct(":"),
126 ct.packetlist_value(other))
127 return "%s%s%s%s%s" % (ct.punct("<"),
128 ct.packetlist_name(self.listname),
129 ct.punct(":"),
130 s,
131 ct.punct(">"))
133 def __getstate__(self):
134 # type: () -> Dict[str, Any]
135 """
136 Creates a basic representation of the instance, used in
137 conjunction with __setstate__() e.g. by pickle
139 :returns: dict representing this instance
140 """
141 state = {
142 'res': self.res,
143 'stats': self.stats,
144 'listname': self.listname
145 }
146 return state
148 def __setstate__(self, state):
149 # type: (Dict[str, Any]) -> None
150 """
151 Sets instance attributes to values given by state, used in
152 conjunction with __getstate__() e.g. by pickle
154 :param state: dict representing this instance
155 """
156 self.res = state['res']
157 self.stats = state['stats']
158 self.listname = state['listname']
160 def __iter__(self):
161 # type: () -> Iterator[_Inner]
162 return self.res.__iter__()
164 def __getattr__(self, attr):
165 # type: (str) -> Any
166 return getattr(self.res, attr)
168 def __getitem__(self, item):
169 # type: (Any) -> Any
170 if issubtype(item, BasePacket):
171 return self.__class__([x for x in self.res if item in self._elt2pkt(x)], # noqa: E501
172 name="%s from %s" % (item.__name__, self.listname)) # noqa: E501
173 if isinstance(item, slice):
174 return self.__class__(self.res.__getitem__(item),
175 name="mod %s" % self.listname)
176 return self.res.__getitem__(item)
178 _T = TypeVar('_T', 'SndRcvList', 'PacketList')
180 # Hinting hack: type self
181 def __add__(self, # type: _PacketList._T # type: ignore
182 other # type: _PacketList._T
183 ):
184 # type: (...) -> _PacketList._T
185 return self.__class__(
186 self.res + other.res,
187 name="%s+%s" % (
188 self.listname,
189 other.listname
190 )
191 )
193 def summary(self,
194 prn=None, # type: Optional[Callable[..., Any]]
195 lfilter=None # type: Optional[Callable[..., bool]]
196 ):
197 # type: (...) -> None
198 """prints a summary of each packet
200 :param prn: function to apply to each packet instead of
201 lambda x:x.summary()
202 :param lfilter: truth function to apply to each packet to decide
203 whether it will be displayed
204 """
205 for r in self.res:
206 if lfilter is not None:
207 if not lfilter(*r):
208 continue
209 if prn is None:
210 print(self._elt2sum(r))
211 else:
212 print(prn(*r))
214 def nsummary(self,
215 prn=None, # type: Optional[Callable[..., Any]]
216 lfilter=None # type: Optional[Callable[..., bool]]
217 ):
218 # type: (...) -> None
219 """prints a summary of each packet with the packet's number
221 :param prn: function to apply to each packet instead of
222 lambda x:x.summary()
223 :param lfilter: truth function to apply to each packet to decide
224 whether it will be displayed
225 """
226 for i, res in enumerate(self.res):
227 if lfilter is not None:
228 if not lfilter(*res):
229 continue
230 print(conf.color_theme.id(i, fmt="%04i"), end=' ')
231 if prn is None:
232 print(self._elt2sum(res))
233 else:
234 print(prn(*res))
236 def show(self, *args, **kargs):
237 # type: (*Any, **Any) -> None
238 """Best way to display the packet list. Defaults to nsummary() method""" # noqa: E501
239 return self.nsummary(*args, **kargs)
241 def filter(self, func):
242 # type: (Callable[..., bool]) -> _PacketList[_Inner]
243 """Returns a packet list filtered by a truth function. This truth
244 function has to take a packet as the only argument and return
245 a boolean value.
246 """
247 return self.__class__([x for x in self.res if func(*x)],
248 name="filtered %s" % self.listname)
250 def make_table(self, *args, **kargs):
251 # type: (Any, Any) -> Optional[str]
252 """Prints a table using a function that returns for each packet its head column value, head row value and displayed value # noqa: E501
253 ex: p.make_table(lambda x:(x[IP].dst, x[TCP].dport, x[TCP].sprintf("%flags%")) """ # noqa: E501
254 return make_table(self.res, *args, **kargs)
256 def make_lined_table(self, *args, **kargs):
257 # type: (Any, Any) -> Optional[str]
258 """Same as make_table, but print a table with lines"""
259 return make_lined_table(self.res, *args, **kargs)
261 def make_tex_table(self, *args, **kargs):
262 # type: (Any, Any) -> Optional[str]
263 """Same as make_table, but print a table with LaTeX syntax"""
264 return make_tex_table(self.res, *args, **kargs)
266 def plot(self,
267 f, # type: Callable[..., Any]
268 lfilter=None, # type: Optional[Callable[..., bool]]
269 plot_xy=False, # type: bool
270 **kargs # type: Any
271 ):
272 # type: (...) -> Line2D
273 """Applies a function to each packet to get a value that will be plotted
274 with matplotlib. A list of matplotlib.lines.Line2D is returned.
276 lfilter: a truth function that decides whether a packet must be plotted
277 """
278 # Defer imports of matplotlib until its needed
279 # because it has a heavy dep chain
280 from scapy.libs.matplot import (
281 plt,
282 MATPLOTLIB_INLINED,
283 MATPLOTLIB_DEFAULT_PLOT_KARGS
284 )
286 # Get the list of packets
287 if lfilter is None:
288 lst_pkts = [f(*e) for e in self.res]
289 else:
290 lst_pkts = [f(*e) for e in self.res if lfilter(*e)]
292 # Mimic the default gnuplot output
293 if kargs == {}:
294 kargs = MATPLOTLIB_DEFAULT_PLOT_KARGS
295 if plot_xy:
296 lines = plt.plot(*zip(*lst_pkts), **kargs)
297 else:
298 lines = plt.plot(lst_pkts, **kargs)
300 # Call show() if matplotlib is not inlined
301 if not MATPLOTLIB_INLINED:
302 plt.show()
304 return lines
306 def diffplot(self,
307 f, # type: Callable[..., Any]
308 delay=1, # type: int
309 lfilter=None, # type: Optional[Callable[..., bool]]
310 **kargs # type: Any
311 ):
312 # type: (...) -> Line2D
313 """diffplot(f, delay=1, lfilter=None)
314 Applies a function to couples (l[i],l[i+delay])
316 A list of matplotlib.lines.Line2D is returned.
317 """
318 # Defer imports of matplotlib until its needed
319 # because it has a heavy dep chain
320 from scapy.libs.matplot import (
321 plt,
322 MATPLOTLIB_INLINED,
323 MATPLOTLIB_DEFAULT_PLOT_KARGS
324 )
326 # Get the list of packets
327 if lfilter is None:
328 lst_pkts = [f(self.res[i], self.res[i + 1])
329 for i in range(len(self.res) - delay)]
330 else:
331 lst_pkts = [f(self.res[i], self.res[i + 1])
332 for i in range(len(self.res) - delay)
333 if lfilter(self.res[i])]
335 # Mimic the default gnuplot output
336 if kargs == {}:
337 kargs = MATPLOTLIB_DEFAULT_PLOT_KARGS
338 lines = plt.plot(lst_pkts, **kargs)
340 # Call show() if matplotlib is not inlined
341 if not MATPLOTLIB_INLINED:
342 plt.show()
344 return lines
346 def multiplot(self,
347 f, # type: Callable[..., Any]
348 lfilter=None, # type: Optional[Callable[..., Any]]
349 plot_xy=False, # type: bool
350 **kargs # type: Any
351 ):
352 # type: (...) -> Line2D
353 """Uses a function that returns a label and a value for this label, then
354 plots all the values label by label.
356 A list of matplotlib.lines.Line2D is returned.
357 """
358 # Defer imports of matplotlib until its needed
359 # because it has a heavy dep chain
360 from scapy.libs.matplot import (
361 plt,
362 MATPLOTLIB_INLINED,
363 MATPLOTLIB_DEFAULT_PLOT_KARGS
364 )
366 # Get the list of packets
367 if lfilter is None:
368 lst_pkts = (f(*e) for e in self.res)
369 else:
370 lst_pkts = (f(*e) for e in self.res if lfilter(*e))
372 # Apply the function f to the packets
373 d = {} # type: Dict[str, List[float]]
374 for k, v in lst_pkts:
375 d.setdefault(k, []).append(v)
377 # Mimic the default gnuplot output
378 if not kargs:
379 kargs = MATPLOTLIB_DEFAULT_PLOT_KARGS
381 if plot_xy:
382 lines = [plt.plot(*list(zip(*pl)), **dict(kargs, label=k))
383 for k, pl in d.items()]
384 else:
385 lines = [plt.plot(pl, **dict(kargs, label=k))
386 for k, pl in d.items()]
387 plt.legend(loc="center right", bbox_to_anchor=(1.5, 0.5))
389 # Call show() if matplotlib is not inlined
390 if not MATPLOTLIB_INLINED:
391 plt.show()
393 return lines
395 def rawhexdump(self):
396 # type: () -> None
397 """Prints an hexadecimal dump of each packet in the list"""
398 for p in self:
399 hexdump(self._elt2pkt(p))
401 def hexraw(self, lfilter=None):
402 # type: (Optional[Callable[..., bool]]) -> None
403 """Same as nsummary(), except that if a packet has a Raw layer, it will be hexdumped # noqa: E501
404 lfilter: a truth function that decides whether a packet must be displayed""" # noqa: E501
405 for i, res in enumerate(self.res):
406 p = self._elt2pkt(res)
407 if lfilter is not None and not lfilter(p):
408 continue
409 print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
410 p.sprintf("%.time%"),
411 self._elt2sum(res)))
412 if p.haslayer(conf.raw_layer):
413 hexdump(p.getlayer(conf.raw_layer).load) # type: ignore
415 def hexdump(self, lfilter=None):
416 # type: (Optional[Callable[..., bool]]) -> None
417 """Same as nsummary(), except that packets are also hexdumped
418 lfilter: a truth function that decides whether a packet must be displayed""" # noqa: E501
419 for i, res in enumerate(self.res):
420 p = self._elt2pkt(res)
421 if lfilter is not None and not lfilter(p):
422 continue
423 print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
424 p.sprintf("%.time%"),
425 self._elt2sum(res)))
426 hexdump(p)
428 def padding(self, lfilter=None):
429 # type: (Optional[Callable[..., bool]]) -> None
430 """Same as hexraw(), for Padding layer"""
431 for i, res in enumerate(self.res):
432 p = self._elt2pkt(res)
433 if p.haslayer(conf.padding_layer):
434 if lfilter is None or lfilter(p):
435 print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
436 p.sprintf("%.time%"),
437 self._elt2sum(res)))
438 hexdump(
439 p.getlayer(conf.padding_layer).load # type: ignore
440 )
442 def nzpadding(self, lfilter=None):
443 # type: (Optional[Callable[..., bool]]) -> None
444 """Same as padding() but only non null padding"""
445 for i, res in enumerate(self.res):
446 p = self._elt2pkt(res)
447 if p.haslayer(conf.padding_layer):
448 pad = p.getlayer(conf.padding_layer).load # type: ignore
449 if pad == pad[:1] * len(pad):
450 continue
451 if lfilter is None or lfilter(p):
452 print("%s %s %s" % (conf.color_theme.id(i, fmt="%04i"),
453 p.sprintf("%.time%"),
454 self._elt2sum(res)))
455 hexdump(
456 p.getlayer(conf.padding_layer).load # type: ignore
457 )
459 def conversations(self,
460 getsrcdst=None, # type: Optional[Callable[[Packet], Tuple[Any, ...]]] # noqa: E501
461 **kargs # type: Any
462 ):
463 # type: (...) -> Any
464 """Graphes a conversations between sources and destinations and display it
465 (using graphviz and imagemagick)
467 :param getsrcdst: a function that takes an element of the list and
468 returns the source, the destination and optionally
469 a label. By default, returns the IP source and
470 destination from IP and ARP layers
471 :param type: output type (svg, ps, gif, jpg, etc.), passed to dot's
472 "-T" option
473 :param target: filename or redirect. Defaults pipe to Imagemagick's
474 display program
475 :param prog: which graphviz program to use
476 """
477 if getsrcdst is None:
478 def _getsrcdst(pkt):
479 # type: (Packet) -> Tuple[str, str]
480 """Extract src and dst addresses"""
481 if 'IP' in pkt:
482 return (pkt['IP'].src, pkt['IP'].dst)
483 if 'IPv6' in pkt:
484 return (pkt['IPv6'].src, pkt['IPv6'].dst)
485 if 'ARP' in pkt:
486 return (pkt['ARP'].psrc, pkt['ARP'].pdst)
487 raise TypeError()
488 getsrcdst = _getsrcdst
489 conv = {} # type: Dict[Tuple[Any, ...], Any]
490 for elt in self.res:
491 p = self._elt2pkt(elt)
492 try:
493 c = getsrcdst(p)
494 except Exception:
495 # No warning here: it's OK that getsrcdst() raises an
496 # exception, since it might be, for example, a
497 # function that expects a specific layer in each
498 # packet. The try/except approach is faster and
499 # considered more Pythonic than adding tests.
500 continue
501 if len(c) == 3:
502 conv.setdefault(c[:2], set()).add(c[2])
503 else:
504 conv[c] = conv.get(c, 0) + 1
505 gr = 'digraph "conv" {\n'
506 for (s, d), l in conv.items():
507 gr += '\t "%s" -> "%s" [label="%s"]\n' % (
508 graphviz_escape(s),
509 graphviz_escape(d),
510 graphviz_escape(
511 ', '.join(str(x) for x in l) if isinstance(l, set) else l
512 ),
513 )
514 gr += "}\n"
515 return do_graph(gr, **kargs)
517 def afterglow(self,
518 src=None, # type: Optional[Callable[[_Inner], Any]]
519 event=None, # type: Optional[Callable[[_Inner], Any]]
520 dst=None, # type: Optional[Callable[[_Inner], Any]]
521 **kargs # type: Any
522 ):
523 # type: (...) -> Any
524 """Experimental clone attempt of http://sourceforge.net/projects/afterglow
525 each datum is reduced as src -> event -> dst and the data are graphed.
526 by default we have IP.src -> IP.dport -> IP.dst"""
527 if src is None:
528 src = lambda *x: x[0]['IP'].src
529 if event is None:
530 event = lambda *x: x[0]['IP'].dport
531 if dst is None:
532 dst = lambda *x: x[0]['IP'].dst
533 sl = {} # type: Dict[Any, Tuple[Union[float, int], List[Any]]]
534 el = {} # type: Dict[Any, Tuple[Union[float, int], List[Any]]]
535 dl = {} # type: Dict[Any, int]
536 for i in self.res:
537 try:
538 s, e, d = src(i), event(i), dst(i)
539 if s in sl:
540 n, lst = sl[s]
541 n += 1
542 if e not in lst:
543 lst.append(e)
544 sl[s] = (n, lst)
545 else:
546 sl[s] = (1, [e])
547 if e in el:
548 n, lst = el[e]
549 n += 1
550 if d not in lst:
551 lst.append(d)
552 el[e] = (n, lst)
553 else:
554 el[e] = (1, [d])
555 dl[d] = dl.get(d, 0) + 1
556 except Exception:
557 continue
559 def minmax(x):
560 # type: (Any) -> Tuple[int, int]
561 m, M = reduce(lambda a, b: (min(a[0], b[0]), max(a[1], b[1])),
562 ((a, a) for a in x))
563 if m == M:
564 m = 0
565 if M == 0:
566 M = 1
567 return m, M
569 mins, maxs = minmax(x for x, _ in sl.values())
570 mine, maxe = minmax(x for x, _ in el.values())
571 mind, maxd = minmax(dl.values())
573 gr = 'digraph "afterglow" {\n\tedge [len=2.5];\n'
575 gr += "# src nodes\n"
576 for s in sl:
577 n, _ = sl[s]
578 n = 1 + float(n - mins) / (maxs - mins)
579 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
580 gr += "# event nodes\n"
581 for e in el:
582 n, _ = el[e]
583 n = 1 + float(n - mine) / (maxe - mine)
584 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
585 for d in dl:
586 n = dl[d]
587 n = 1 + float(n - mind) / (maxd - mind)
588 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
590 gr += "###\n"
591 for s in sl:
592 n, lst1 = sl[s]
593 for e in lst1:
594 gr += ' "src.%s" -> "evt.%s";\n' % (repr(s), repr(e))
595 for e in el:
596 n, lst2 = el[e]
597 for d in lst2:
598 gr += ' "evt.%s" -> "dst.%s";\n' % (repr(e), repr(d))
600 gr += "}"
601 return do_graph(gr, **kargs)
603 def canvas_dump(self, layer_shift=0, rebuild=1):
604 # type: (int, int) -> 'pyx.canvas.canvas'
605 d = pyx.document.document()
606 len_res = len(self.res)
607 for i, res in enumerate(self.res):
608 c = self._elt2pkt(res).canvas_dump(layer_shift=layer_shift,
609 rebuild=rebuild)
610 cbb = c.bbox()
611 c.text(cbb.left(), cbb.top() + 1, r"\font\cmssfont=cmss12\cmssfont{Frame %i/%i}" % (i, len_res), [pyx.text.size.LARGE]) # noqa: E501
612 if conf.verb >= 2:
613 os.write(1, b".")
614 d.append(pyx.document.page(c, paperformat=pyx.document.paperformat.A4, # noqa: E501
615 margin=1 * pyx.unit.t_cm,
616 fittosize=1))
617 return d
619 def sessions(
620 self,
621 session_extractor=None # type: Optional[Callable[[Packet], str]]
622 ):
623 # type: (...) -> Dict[str, _PacketList[_Inner]]
624 if session_extractor is None:
625 def _session_extractor(p):
626 # type: (Packet) -> str
627 """Extract sessions from packets"""
628 if 'Ether' in p:
629 if 'IP' in p or 'IPv6' in p:
630 ip_src_fmt = "{IP:%IP.src%}{IPv6:%IPv6.src%}"
631 ip_dst_fmt = "{IP:%IP.dst%}{IPv6:%IPv6.dst%}"
632 addr_fmt = (ip_src_fmt, ip_dst_fmt)
633 if 'TCP' in p:
634 fmt = "TCP {}:%r,TCP.sport% > {}:%r,TCP.dport%"
635 elif 'UDP' in p:
636 fmt = "UDP {}:%r,UDP.sport% > {}:%r,UDP.dport%"
637 elif 'ICMP' in p:
638 fmt = "ICMP {} > {} type=%r,ICMP.type% code=%r," \
639 "ICMP.code% id=%ICMP.id%"
640 elif 'ICMPv6' in p:
641 fmt = "ICMPv6 {} > {} type=%r,ICMPv6.type% " \
642 "code=%r,ICMPv6.code%"
643 elif 'IPv6' in p:
644 fmt = "IPv6 {} > {} nh=%IPv6.nh%"
645 else:
646 fmt = "IP {} > {} proto=%IP.proto%"
647 return p.sprintf(fmt.format(*addr_fmt))
648 elif 'ARP' in p:
649 return p.sprintf("ARP %ARP.psrc% > %ARP.pdst%")
650 else:
651 return p.sprintf("Ethernet type=%04xr,Ether.type%")
652 return "Other"
653 session_extractor = _session_extractor
654 sessions = defaultdict(self.__class__) # type: DefaultDict[str, _PacketList[_Inner]] # noqa: E501
655 for p in self.res:
656 sess = session_extractor(
657 self._elt2pkt(p)
658 )
659 sessions[sess].append(p)
660 return dict(sessions)
662 def replace(self, *args, **kargs):
663 # type: (Any, Any) -> PacketList
664 """
665 lst.replace(<field>,[<oldvalue>,]<newvalue>)
666 lst.replace( (fld,[ov],nv),(fld,[ov,]nv),...)
667 if ov is None, all values are replaced
668 ex:
669 lst.replace( IP.src, "192.168.1.1", "10.0.0.1" )
670 lst.replace( IP.ttl, 64 )
671 lst.replace( (IP.ttl, 64), (TCP.sport, 666, 777), )
672 """
673 delete_checksums = kargs.get("delete_checksums", False)
674 x = PacketList(name="Replaced %s" % self.listname)
675 if not isinstance(args[0], tuple):
676 args = (args,)
677 for _p in self.res:
678 p = self._elt2pkt(_p)
679 copied = False
680 for scheme in args:
681 fld = scheme[0]
682 old = scheme[1] # not used if len(scheme) == 2
683 new = scheme[-1]
684 for o in fld.owners:
685 if o in p:
686 if len(scheme) == 2 or p[o].getfieldval(fld.name) == old: # noqa: E501
687 if not copied:
688 p = p.copy()
689 if delete_checksums:
690 p.delete_checksums()
691 copied = True
692 setattr(p[o], fld.name, new)
693 x.append(p)
694 return x
696 def getlayer(self, cls, # type: Packet
697 nb=None, # type: Optional[int]
698 flt=None, # type: Optional[Dict[str, Any]]
699 name=None, # type: Optional[str]
700 stats=None # type: Optional[List[Type[Packet]]]
701 ):
702 # type: (...) -> PacketList
703 """Returns the packet list from a given layer.
705 See ``Packet.getlayer`` for more info.
707 :param cls: search for a layer that is an instance of ``cls``
708 :type cls: Type[scapy.packet.Packet]
710 :param nb: return the nb^th layer that is an instance of ``cls``
711 :type nb: Optional[int]
713 :param flt: filter parameters for ``Packet.getlayer``
714 :type flt: Optional[Dict[str, Any]]
716 :param name: optional name for the new PacketList
717 :type name: Optional[str]
719 :param stats: optional list of protocols to give stats on; if not
720 specified, inherits from this PacketList.
721 :type stats: Optional[List[Type[scapy.packet.Packet]]]
722 :rtype: scapy.plist.PacketList
723 """
724 if name is None:
725 name = "{} layer {}".format(self.listname, cls.__name__)
726 if stats is None:
727 stats = self.stats
729 getlayer_arg = {} # type: Dict[str, Any]
730 if flt is not None:
731 getlayer_arg.update(flt)
732 getlayer_arg['cls'] = cls
733 if nb is not None:
734 getlayer_arg['nb'] = nb
736 # Only return non-None getlayer results
737 return PacketList([
738 pc for pc in (
739 self._elt2pkt(p).getlayer(**getlayer_arg) for p in self.res
740 ) if pc is not None],
741 name, stats
742 )
745class PacketList(_PacketList[Packet],
746 BasePacketList[Packet],
747 _CanvasDumpExtended):
748 def sr(self, multi=False, lookahead=None):
749 # type: (bool, Optional[int]) -> Tuple[SndRcvList, PacketList]
750 """
751 Matches packets in the list
753 :param multi: True if a packet can have multiple answers
754 :param lookahead: Maximum number of packets between packet and answer.
755 If 0 or None, full remaining list is
756 scanned for answers
757 :return: ( (matched couples), (unmatched packets) )
758 """
759 remain = self.res[:]
760 sr = [] # type: List[QueryAnswer]
761 i = 0
762 if lookahead is None or lookahead == 0:
763 lookahead = len(remain)
764 while i < len(remain):
765 s = remain[i]
766 j = i
767 while j < min(lookahead + i, len(remain) - 1):
768 j += 1
769 r = remain[j]
770 if r.answers(s):
771 sr.append(QueryAnswer(s, r))
772 if multi:
773 remain[i]._answered = 1
774 remain[j]._answered = 2
775 continue
776 del remain[j]
777 del remain[i]
778 i -= 1
779 break
780 i += 1
781 if multi:
782 remain = [x for x in remain if not hasattr(x, "_answered")]
783 return SndRcvList(sr), PacketList(remain)
786_PacketIterable = Union[
787 Sequence[Packet],
788 Packet,
789 SetGen[Packet],
790 _PacketList[Packet]
791]
794class SndRcvList(_PacketList[QueryAnswer],
795 BasePacketList[QueryAnswer],
796 _CanvasDumpExtended):
797 __slots__ = [] # type: List[str]
799 def __init__(self,
800 res=None, # type: Optional[Union[_PacketList[QueryAnswer], List[QueryAnswer]]] # noqa: E501
801 name="Results", # type: str
802 stats=None # type: Optional[List[Type[Packet]]]
803 ):
804 # type: (...) -> None
805 super(SndRcvList, self).__init__(res, name, stats)
807 def _elt2pkt(self, elt):
808 # type: (QueryAnswer) -> Packet
809 return elt[1]
811 def _elt2sum(self, elt):
812 # type: (QueryAnswer) -> str
813 return "%s ==> %s" % (elt[0].summary(), elt[1].summary())