Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/packet.py: 47%
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"""
7Packet class
9Provides:
10 - the default Packet classes
11 - binding mechanisms
12 - fuzz() method
13 - exploration methods: explore() / ls()
14"""
16from collections import defaultdict
18import json
19import re
20import time
21import itertools
22import copy
23import types
24import warnings
26from scapy.fields import (
27 AnyField,
28 BitField,
29 ConditionalField,
30 Emph,
31 EnumField,
32 Field,
33 FlagsField,
34 FlagValue,
35 MultiEnumField,
36 MultipleTypeField,
37 PadField,
38 PacketListField,
39 RawVal,
40 StrField,
41)
42from scapy.config import conf, _version_checker
43from scapy.compat import raw, bytes_encode
44from scapy.base_classes import BasePacket, Gen, SetGen, Packet_metaclass, \
45 _CanvasDumpExtended
46from scapy.interfaces import _GlobInterfaceType
47from scapy.volatile import RandField, VolatileValue
48from scapy.utils import import_hexcap, tex_escape, colgen, issubtype, \
49 pretty_list, EDecimal
50from scapy.error import Scapy_Exception, log_runtime, warning
51from scapy.libs.test_pyx import PYX
53# Typing imports
54from typing import (
55 Any,
56 Callable,
57 ClassVar,
58 Dict,
59 Iterator,
60 List,
61 NoReturn,
62 Optional,
63 Set,
64 Tuple,
65 Type,
66 TypeVar,
67 Union,
68 Sequence,
69 cast,
70)
71from scapy.compat import Self
73try:
74 import pyx
75except ImportError:
76 pass
79_T = TypeVar("_T", Dict[str, Any], Optional[Dict[str, Any]])
82class Packet(
83 BasePacket,
84 _CanvasDumpExtended,
85 metaclass=Packet_metaclass
86):
87 __slots__ = [
88 "time", "sent_time", "name",
89 "default_fields", "fields", "fieldtype",
90 "overload_fields", "overloaded_fields",
91 "packetfields",
92 "original", "explicit", "raw_packet_cache",
93 "raw_packet_cache_fields", "_pkt", "post_transforms",
94 "stop_dissection_after",
95 # then payload, underlayer and parent
96 "payload", "underlayer", "parent",
97 "name",
98 # used for sr()
99 "_answered",
100 # used when sniffing
101 "direction", "sniffed_on",
102 # handle snaplen Vs real length
103 "wirelen",
104 "comments",
105 "process_information"
106 ]
107 # Types of the __slots__ above: __init__ sets them through
108 # object.__setattr__, which leaves nowhere to put an inline type comment.
109 time: Union[EDecimal, float]
110 sent_time: Union[EDecimal, float, None]
111 default_fields: Dict[str, Any]
112 fields: Dict[str, Any]
113 fieldtype: Dict[str, AnyField]
114 overloaded_fields: Dict[str, Any]
115 packetfields: List[AnyField]
116 original: bytes
117 explicit: int
118 raw_packet_cache: Optional[bytes]
119 raw_packet_cache_fields: Optional[Dict[str, Any]]
120 stop_dissection_after: Optional[Type['Packet']]
121 payload: 'Packet'
122 underlayer: Optional['Packet']
123 parent: Optional['Packet']
124 direction: Optional[int]
125 sniffed_on: Optional[_GlobInterfaceType]
126 wirelen: Optional[int]
127 comments: Optional[List[bytes]]
128 process_information: Optional[Dict[str, Any]]
129 name = None # type: Optional[str]
130 fields_desc = [] # type: ClassVar[List[AnyField]]
131 deprecated_fields = {} # type: Dict[str, Tuple[str, str]]
132 overload_fields = {} # type: Dict[Type[Packet], Dict[str, Any]]
133 payload_guess = [] # type: List[Tuple[Dict[str, Any], Type[Packet]]]
134 show_indent = 1
135 show_summary = True
136 match_subclass = False
137 class_dont_cache = {} # type: Dict[Type[Packet], bool]
138 class_packetfields = {} # type: Dict[Type[Packet], Any]
139 class_default_fields = {} # type: Dict[Type[Packet], Dict[str, Any]]
140 class_default_fields_ref = {} # type: Dict[Type[Packet], List[str]]
141 class_fieldtype = {} # type: Dict[Type[Packet], Dict[str, AnyField]] # noqa: E501
143 @classmethod
144 def from_hexcap(cls):
145 # type: (Type[Packet]) -> Packet
146 return cls(import_hexcap())
148 @classmethod
149 def upper_bonds(self):
150 # type: () -> None
151 for fval, upper in self.payload_guess:
152 print(
153 "%-20s %s" % (
154 upper.__name__,
155 ", ".join("%-12s" % ("%s=%r" % i) for i in fval.items()),
156 )
157 )
159 @classmethod
160 def lower_bonds(self):
161 # type: () -> None
162 for lower, fval in self._overload_fields.items():
163 print(
164 "%-20s %s" % (
165 lower.__name__,
166 ", ".join("%-12s" % ("%s=%r" % i) for i in fval.items()),
167 )
168 )
170 def __init__(self,
171 _pkt=b"", # type: Union[bytes, bytearray]
172 post_transform=None, # type: Any
173 _internal=0, # type: int
174 _underlayer=None, # type: Optional[Packet]
175 _parent=None, # type: Optional[Packet]
176 stop_dissection_after=None, # type: Optional[Type[Packet]]
177 **fields # type: Any
178 ):
179 # type: (...) -> None
180 # Every attribute set below is a __slots__ member, so bypass
181 # __setattr__: resolving field names there costs more than the
182 # assignment itself, and this runs for every dissected layer.
183 object.__setattr__(self, "time", 0.0 if _internal else time.time())
184 object.__setattr__(self, "sent_time", None)
185 object.__setattr__(self, "name", self.__class__.__name__
186 if self._name is None else self._name)
187 object.__setattr__(self, "default_fields", {})
188 object.__setattr__(self, "overload_fields", self._overload_fields)
189 object.__setattr__(self, "overloaded_fields", {})
190 object.__setattr__(self, "fields", {})
191 object.__setattr__(self, "fieldtype", {})
192 object.__setattr__(self, "packetfields", [])
193 object.__setattr__(self, "payload", NoPayload())
194 self.init_fields(bool(_pkt))
195 object.__setattr__(self, "underlayer", _underlayer)
196 object.__setattr__(self, "parent", _parent)
197 if isinstance(_pkt, bytearray):
198 _pkt = bytes(_pkt)
199 object.__setattr__(self, "original", _pkt)
200 object.__setattr__(self, "explicit", 0)
201 object.__setattr__(self, "raw_packet_cache", None)
202 object.__setattr__(self, "raw_packet_cache_fields", None)
203 object.__setattr__(self, "wirelen", None)
204 object.__setattr__(self, "direction", None)
205 object.__setattr__(self, "sniffed_on", None)
206 object.__setattr__(self, "comments", None)
207 object.__setattr__(self, "process_information", None)
208 object.__setattr__(self, "stop_dissection_after", stop_dissection_after)
209 if _pkt:
210 self.dissect(_pkt)
211 if not _internal:
212 self.dissection_done(self)
213 # We use this strange initialization so that the fields
214 # are initialized in their declaration order.
215 # It is required to always support MultipleTypeField
216 for field in self.fields_desc:
217 fname = field.name
218 try:
219 value = fields.pop(fname)
220 except KeyError:
221 continue
222 self.fields[fname] = value if isinstance(value, RawVal) else \
223 self.get_field(fname).any2i(self, value)
224 # The remaining fields are unknown
225 for fname in fields:
226 if fname in self.deprecated_fields:
227 # Resolve deprecated fields
228 value = fields[fname]
229 fname = self._resolve_alias(fname)
230 self.fields[fname] = value if isinstance(value, RawVal) else \
231 self.get_field(fname).any2i(self, value)
232 continue
233 raise AttributeError(fname)
234 if isinstance(post_transform, list):
235 self.post_transforms = post_transform
236 elif post_transform is None:
237 self.post_transforms = []
238 else:
239 self.post_transforms = [post_transform]
241 _PickleType = Tuple[
242 Union[EDecimal, float],
243 Optional[Union[EDecimal, float, None]],
244 Optional[int],
245 Optional[_GlobInterfaceType],
246 Optional[int],
247 Optional[bytes],
248 ]
249 _PickleStateType = Union[_PickleType, Dict[str, Any]]
251 @property
252 def comment(self):
253 # type: () -> Optional[bytes]
254 """Get the comment of the packet"""
255 if self.comments and len(self.comments):
256 return self.comments[0]
257 return None
259 @comment.setter
260 def comment(self, value):
261 # type: (Optional[bytes]) -> None
262 """
263 Set the comment of the packet.
264 If value is None, it will clear the comments.
265 """
266 if value is not None:
267 self.comments = [value]
268 else:
269 self.comments = None
271 @classmethod
272 def _rebuild_pkt(cls, raw_packet):
273 # type: (Type[Packet], bytes) -> Packet
274 """Helper used by pickle to reconstruct Packet from raw bytes."""
275 return cls(raw_packet)
277 def __reduce__(self):
278 # type: () -> Tuple[Any, ...]
279 """Used by pickling methods"""
280 state = {
281 "pickle_state_version": 2,
282 "time": self.time,
283 "sent_time": self.sent_time,
284 "direction": self.direction,
285 "sniffed_on": self.sniffed_on,
286 "wirelen": self.wirelen,
287 # Keep both keys for compatibility with historical/transition code.
288 "comment": self.comment,
289 "comments": self.comments,
290 }
291 extra_slots = {}
292 for attr in type(self).__all_slots__ - set(Packet.__slots__):
293 if hasattr(self, attr):
294 extra_slots[attr] = getattr(self, attr)
295 if extra_slots:
296 state["extra_slots"] = extra_slots # type: ignore
297 return (type(self)._rebuild_pkt, (self.build(),), state)
299 def __setstate__(self, state):
300 # type: (Packet._PickleStateType) -> Packet
301 """Rebuild state using pickable methods"""
302 # Legacy format: tuple produced by older Packet.__reduce__.
303 if isinstance(state, tuple):
304 self.time = state[0]
305 self.sent_time = state[1]
306 self.direction = state[2]
307 self.sniffed_on = state[3]
308 self.wirelen = state[4]
309 self.comment = state[5]
310 return self
312 # New format: versioned dict metadata.
313 self.time = state.get("time", self.time)
314 self.sent_time = state.get("sent_time", self.sent_time)
315 self.direction = state.get("direction", self.direction)
316 self.sniffed_on = state.get("sniffed_on", self.sniffed_on)
317 self.wirelen = state.get("wirelen", self.wirelen)
319 if "comments" in state:
320 self.comments = state["comments"]
321 elif "comment" in state:
322 self.comment = state["comment"]
324 extra_slots = state.get("extra_slots", {})
325 if isinstance(extra_slots, dict):
326 for attr, value in extra_slots.items():
327 # Only restore known subclass slots; ignore stale/unknown entries.
328 if attr in type(self).__all_slots__ and attr not in Packet.__slots__:
329 try:
330 setattr(self, attr, value)
331 except AttributeError:
332 pass
333 return self
335 def __deepcopy__(self,
336 memo, # type: Any
337 ):
338 # type: (...) -> Packet
339 """Used by copy.deepcopy"""
340 return self.copy()
342 def init_fields(self, for_dissect_only=False):
343 # type: (bool) -> None
344 """
345 Initialize each fields of the fields_desc dict
346 """
348 if self.class_dont_cache.get(self.__class__, False):
349 self.do_init_fields(self.fields_desc)
350 else:
351 self.do_init_cached_fields(for_dissect_only=for_dissect_only)
353 def do_init_fields(self,
354 flist, # type: Sequence[AnyField]
355 ):
356 # type: (...) -> None
357 """
358 Initialize each fields of the fields_desc dict
359 """
360 default_fields = {}
361 for f in flist:
362 default_fields[f.name] = copy.deepcopy(f.default)
363 self.fieldtype[f.name] = f
364 if f.holds_packets:
365 self.packetfields.append(f)
366 # We set default_fields last to avoid race issues
367 self.default_fields = default_fields
369 def do_init_cached_fields(self, for_dissect_only=False):
370 # type: (bool) -> None
371 """
372 Initialize each fields of the fields_desc dict, or use the cached
373 fields information
374 """
376 cls_name = self.__class__
378 # Build the fields information
379 default_fields = Packet.class_default_fields.get(cls_name)
380 if default_fields is None:
381 self.prepare_cached_fields(self.fields_desc)
382 default_fields = Packet.class_default_fields.get(cls_name)
384 # Use fields information from cache
385 if default_fields:
386 self.default_fields = default_fields
387 self.fieldtype = Packet.class_fieldtype[cls_name]
388 self.packetfields = Packet.class_packetfields[cls_name]
390 # Optimization: no need for references when only dissecting.
391 if for_dissect_only:
392 return
394 # Deepcopy default references
395 for fname in Packet.class_default_fields_ref[cls_name]:
396 value = self.default_fields[fname]
397 try:
398 self.fields[fname] = value.copy()
399 except AttributeError:
400 # Python 2.7 - list only
401 self.fields[fname] = value[:]
403 def prepare_cached_fields(self, flist):
404 # type: (Sequence[AnyField]) -> None
405 """
406 Prepare the cached fields of the fields_desc dict
407 """
409 cls_name = self.__class__
411 # Fields cache initialization
412 if not flist:
413 return
415 class_default_fields = dict()
416 class_default_fields_ref = list()
417 class_fieldtype = dict()
418 class_packetfields = list()
420 # Fields initialization
421 for f in flist:
422 if isinstance(f, MultipleTypeField):
423 # Abort
424 self.class_dont_cache[cls_name] = True
425 self.do_init_fields(self.fields_desc)
426 return
428 class_default_fields[f.name] = copy.deepcopy(f.default)
429 class_fieldtype[f.name] = f
430 if f.holds_packets:
431 class_packetfields.append(f)
433 # Remember references
434 if isinstance(f.default, (list, dict, set, RandField, Packet)):
435 class_default_fields_ref.append(f.name)
437 # Apply
438 Packet.class_default_fields_ref[cls_name] = class_default_fields_ref
439 Packet.class_fieldtype[cls_name] = class_fieldtype
440 Packet.class_packetfields[cls_name] = class_packetfields
441 # Last to avoid racing issues
442 Packet.class_default_fields[cls_name] = class_default_fields
444 def dissection_done(self, pkt):
445 # type: (Packet) -> None
446 """DEV: will be called after a dissection is completed"""
447 self.post_dissection(pkt)
448 self.payload.dissection_done(pkt)
450 def post_dissection(self, pkt):
451 # type: (Packet) -> None
452 """DEV: is called after the dissection of the whole packet"""
453 pass
455 def get_field(self, fld):
456 # type: (str) -> AnyField
457 """DEV: returns the field instance from the name of the field"""
458 return self.fieldtype[fld]
460 def add_payload(self, payload):
461 # type: (Union[Packet, bytes]) -> None
462 if payload is None:
463 return
464 elif not isinstance(self.payload, NoPayload):
465 self.payload.add_payload(payload)
466 else:
467 if isinstance(payload, Packet):
468 self.payload = payload
469 payload.add_underlayer(self)
470 for t in self.aliastypes:
471 if t in payload.overload_fields:
472 self.overloaded_fields = payload.overload_fields[t]
473 break
474 elif isinstance(payload, (bytes, str, bytearray, memoryview)):
475 self.payload = conf.raw_layer(load=bytes_encode(payload))
476 else:
477 raise TypeError("payload must be 'Packet', 'bytes', 'str', 'bytearray', or 'memoryview', not [%s]" % repr(payload)) # noqa: E501
479 def remove_payload(self):
480 # type: () -> None
481 self.payload.remove_underlayer(self)
482 self.payload = NoPayload()
483 self.overloaded_fields = {}
485 def add_underlayer(self, underlayer):
486 # type: (Packet) -> None
487 self.underlayer = underlayer
489 def remove_underlayer(self, other):
490 # type: (Packet) -> None
491 self.underlayer = None
493 def add_parent(self, parent):
494 # type: (Packet) -> None
495 """Set packet parent.
496 When packet is an element in PacketListField, parent field would
497 point to the list owner packet."""
498 self.parent = parent
500 def remove_parent(self, other):
501 # type: (Packet) -> None
502 """Remove packet parent.
503 When packet is an element in PacketListField, parent field would
504 point to the list owner packet."""
505 self.parent = None
507 def copy(self) -> Self:
508 """Returns a deep copy of the instance."""
509 clone = self.__class__()
510 clone.fields = self.copy_fields_dict(self.fields)
511 clone.default_fields = self.copy_fields_dict(self.default_fields)
512 clone.overloaded_fields = self.overloaded_fields.copy()
513 clone.underlayer = self.underlayer
514 clone.parent = self.parent
515 clone.explicit = self.explicit
516 clone.raw_packet_cache = self.raw_packet_cache
517 clone.raw_packet_cache_fields = self.copy_fields_dict(
518 self.raw_packet_cache_fields
519 )
520 clone.wirelen = self.wirelen
521 clone.post_transforms = self.post_transforms[:]
522 clone.payload = self.payload.copy()
523 clone.payload.add_underlayer(clone)
524 clone.time = self.time
525 clone.comments = self.comments
526 clone.direction = self.direction
527 clone.sniffed_on = self.sniffed_on
528 return clone
530 def _resolve_alias(self, attr):
531 # type: (str) -> str
532 new_attr, version = self.deprecated_fields[attr]
533 warnings.warn(
534 "%s has been deprecated in favor of %s since %s !" % (
535 attr, new_attr, version
536 ), DeprecationWarning
537 )
538 return new_attr
540 def getfieldval(self, attr):
541 # type: (str) -> Any
542 if self.deprecated_fields and attr in self.deprecated_fields:
543 attr = self._resolve_alias(attr)
544 try:
545 return self.fields[attr]
546 except KeyError:
547 pass
548 try:
549 return self.overloaded_fields[attr]
550 except KeyError:
551 pass
552 try:
553 return self.default_fields[attr]
554 except KeyError:
555 pass
556 return self.payload.getfieldval(attr)
558 def getfield_and_val(self, attr):
559 # type: (str) -> Tuple[AnyField, Any]
560 if self.deprecated_fields and attr in self.deprecated_fields:
561 attr = self._resolve_alias(attr)
562 if attr in self.fields:
563 return self.fieldtype[attr], self.fields[attr]
564 if attr in self.overloaded_fields:
565 return self.fieldtype[attr], self.overloaded_fields[attr]
566 if attr in self.default_fields:
567 return self.fieldtype[attr], self.default_fields[attr]
568 raise ValueError
570 def __getattr__(self, attr):
571 # type: (str) -> Any
572 try:
573 fld, v = self.getfield_and_val(attr)
574 except ValueError:
575 return self.payload.__getattr__(attr)
576 if fld is not None:
577 return v if isinstance(v, RawVal) else fld.i2h(self, v)
578 return v
580 def setfieldval(self, attr, val):
581 # type: (str, Any) -> None
582 if self.deprecated_fields and attr in self.deprecated_fields:
583 attr = self._resolve_alias(attr)
584 if attr in self.default_fields:
585 fld = self.get_field(attr)
586 if fld is None:
587 any2i = lambda x, y: y # type: Callable[..., Any]
588 else:
589 any2i = fld.any2i
590 self.fields[attr] = val if isinstance(val, RawVal) else \
591 any2i(self, val)
592 self.explicit = 0
593 self.raw_packet_cache = None
594 self.raw_packet_cache_fields = None
595 self.wirelen = None
596 elif attr == "payload":
597 self.remove_payload()
598 self.add_payload(val)
599 else:
600 self.payload.setfieldval(attr, val)
602 def __setattr__(self, attr, val):
603 # type: (str, Any) -> None
604 if attr in self.__all_slots__:
605 return object.__setattr__(self, attr, val)
606 try:
607 return self.setfieldval(attr, val)
608 except AttributeError:
609 pass
610 return object.__setattr__(self, attr, val)
612 def delfieldval(self, attr):
613 # type: (str) -> None
614 if attr in self.fields:
615 del self.fields[attr]
616 self.explicit = 0 # in case a default value must be explicit
617 self.raw_packet_cache = None
618 self.raw_packet_cache_fields = None
619 self.wirelen = None
620 elif attr in self.default_fields:
621 pass
622 elif attr == "payload":
623 self.remove_payload()
624 else:
625 self.payload.delfieldval(attr)
627 def __delattr__(self, attr):
628 # type: (str) -> None
629 if attr == "payload":
630 return self.remove_payload()
631 if attr in self.__all_slots__:
632 return object.__delattr__(self, attr)
633 try:
634 return self.delfieldval(attr)
635 except AttributeError:
636 pass
637 return object.__delattr__(self, attr)
639 def _superdir(self):
640 # type: () -> Set[str]
641 """
642 Return a list of slots and methods, including those from subclasses.
643 """
644 attrs = set() # type: Set[str]
645 cls = self.__class__
646 if hasattr(cls, '__all_slots__'):
647 attrs.update(cls.__all_slots__)
648 for bcls in cls.__mro__:
649 if hasattr(bcls, '__dict__'):
650 attrs.update(bcls.__dict__)
651 return attrs
653 def __dir__(self):
654 # type: () -> List[str]
655 """
656 Add fields to tab completion list.
657 """
658 return sorted(itertools.chain(self._superdir(), self.default_fields))
660 def __repr__(self):
661 # type: () -> str
662 s = ""
663 ct = conf.color_theme
664 for f in self.fields_desc:
665 if isinstance(f, ConditionalField) and not f._evalcond(self):
666 continue
667 if f.name in self.fields:
668 fval = self.fields[f.name]
669 if isinstance(fval, (list, dict, set)) and len(fval) == 0:
670 continue
671 val = f.i2repr(self, fval)
672 elif f.name in self.overloaded_fields:
673 fover = self.overloaded_fields[f.name]
674 if isinstance(fover, (list, dict, set)) and len(fover) == 0:
675 continue
676 val = f.i2repr(self, fover)
677 else:
678 continue
679 if isinstance(f, Emph) or f in conf.emph:
680 ncol = ct.emph_field_name
681 vcol = ct.emph_field_value
682 else:
683 ncol = ct.field_name
684 vcol = ct.field_value
686 s += " %s%s%s" % (ncol(f.name),
687 ct.punct("="),
688 vcol(val))
689 return "%s%s %s %s%s%s" % (ct.punct("<"),
690 ct.layer_name(self.__class__.__name__),
691 s,
692 ct.punct("|"),
693 repr(self.payload),
694 ct.punct(">"))
696 def __str__(self):
697 # type: () -> str
698 return self.summary()
700 def __bytes__(self):
701 # type: () -> bytes
702 return self.build()
704 def __div__(self, other):
705 # type: (Any) -> Self
706 if isinstance(other, Packet):
707 cloneA = self.copy()
708 cloneB = other.copy()
709 cloneA.add_payload(cloneB)
710 return cloneA
711 elif isinstance(other, (bytes, str, bytearray, memoryview)):
712 return self / conf.raw_layer(load=bytes_encode(other))
713 else:
714 return other.__rdiv__(self) # type: ignore
715 __truediv__ = __div__
717 def __rdiv__(self, other):
718 # type: (Any) -> Packet
719 if isinstance(other, (bytes, str, bytearray, memoryview)):
720 return conf.raw_layer(load=bytes_encode(other)) / self
721 else:
722 raise TypeError
723 __rtruediv__ = __rdiv__
725 def __mul__(self, other):
726 # type: (Any) -> List[Packet]
727 if isinstance(other, int):
728 return [self] * other
729 else:
730 raise TypeError
732 def __rmul__(self, other):
733 # type: (Any) -> List[Packet]
734 return self.__mul__(other)
736 def __nonzero__(self):
737 # type: () -> bool
738 return True
739 __bool__ = __nonzero__
741 def __len__(self):
742 # type: () -> int
743 return len(self.__bytes__())
745 def copy_field_value(self, fieldname, value):
746 # type: (str, Any) -> Any
747 return self.get_field(fieldname).do_copy(value)
749 def copy_fields_dict(self, fields):
750 # type: (_T) -> _T
751 if fields is None:
752 return None
753 fieldtype = self.fieldtype
754 return {fname: fieldtype[fname].do_copy(fval)
755 for fname, fval in fields.items()}
757 def _raw_packet_cache_field_value(self, fld, val, copy=False):
758 # type: (AnyField, Any, bool) -> Optional[Any]
759 """Get a value representative of a mutable field to detect changes"""
760 if fld.holds_packets:
761 # avoid copying whole packets (perf: #GH3894)
762 if fld.islist:
763 if copy:
764 return [
765 (fld.do_copy(x.fields), x.payload.raw_packet_cache)
766 for x in val
767 ]
768 return [
769 (x.fields, x.payload.raw_packet_cache) for x in val
770 ]
771 else:
772 if copy:
773 return (fld.do_copy(val.fields),
774 val.payload.raw_packet_cache)
775 return (val.fields, val.payload.raw_packet_cache)
776 elif fld.islist or fld.ismutable:
777 return fld.do_copy(val) if copy else val
778 return None
780 def clear_cache(self):
781 # type: () -> None
782 """Clear the raw packet cache for the field and all its subfields"""
783 self.raw_packet_cache = None
784 for fname, fval in self.fields.items():
785 fld = self.get_field(fname)
786 if fld.holds_packets:
787 if isinstance(fval, Packet):
788 fval.clear_cache()
789 elif isinstance(fval, list):
790 for fsubval in fval:
791 fsubval.clear_cache()
792 self.payload.clear_cache()
794 def self_build(self):
795 # type: () -> bytes
796 """
797 Create the default layer regarding fields_desc dict
798 """
799 if self.raw_packet_cache is not None and \
800 self.raw_packet_cache_fields is not None:
801 for fname, fval in self.raw_packet_cache_fields.items():
802 fld, val = self.getfield_and_val(fname)
803 if self._raw_packet_cache_field_value(fld, val) != fval:
804 self.raw_packet_cache = None
805 self.raw_packet_cache_fields = None
806 self.wirelen = None
807 break
808 if self.raw_packet_cache is not None:
809 return self.raw_packet_cache
810 p = b""
811 for f in self.fields_desc:
812 val = self.getfieldval(f.name)
813 if isinstance(val, RawVal):
814 p += bytes(val)
815 else:
816 try:
817 p = f.addfield(self, p, val)
818 except Exception as ex:
819 try:
820 ex.args = (
821 "While building field '%s': " % f.name +
822 ex.args[0],
823 ) + ex.args[1:]
824 except (AttributeError, IndexError):
825 pass
826 raise ex
827 return p
829 def do_build_payload(self):
830 # type: () -> bytes
831 """
832 Create the default version of the payload layer
834 :return: a string of payload layer
835 """
836 return self.payload.do_build()
838 def do_build(self):
839 # type: () -> bytes
840 """
841 Create the default version of the layer
843 :return: a string of the packet with the payload
844 """
845 if not self.explicit:
846 self = next(iter(self))
847 pkt = self.self_build()
848 for t in self.post_transforms:
849 pkt = t(pkt)
850 pay = self.do_build_payload()
851 if self.raw_packet_cache is None:
852 return self.post_build(pkt, pay)
853 else:
854 return pkt + pay
856 def build_padding(self):
857 # type: () -> bytes
858 return self.payload.build_padding()
860 def build(self):
861 # type: () -> bytes
862 """
863 Create the current layer
865 :return: string of the packet with the payload
866 """
867 p = self.do_build()
868 p += self.build_padding()
869 p = self.build_done(p)
870 return p
872 def post_build(self, pkt, pay):
873 # type: (bytes, bytes) -> bytes
874 """
875 DEV: called right after the current layer is build.
877 :param str pkt: the current packet (build by self_build function)
878 :param str pay: the packet payload (build by do_build_payload function)
879 :return: a string of the packet with the payload
880 """
881 return pkt + pay
883 def build_done(self, p):
884 # type: (bytes) -> bytes
885 return self.payload.build_done(p)
887 def do_build_ps(self):
888 # type: () -> Tuple[bytes, List[Tuple[Packet, List[Tuple[Field[Any, Any], str, bytes]]]]] # noqa: E501
889 p = b""
890 pl = []
891 q = b""
892 for f in self.fields_desc:
893 if isinstance(f, ConditionalField) and not f._evalcond(self):
894 continue
895 p = f.addfield(self, p, self.getfieldval(f.name))
896 if isinstance(p, bytes):
897 r = p[len(q):]
898 q = p
899 else:
900 r = b""
901 pl.append((f, f.i2repr(self, self.getfieldval(f.name)), r))
903 pkt, lst = self.payload.build_ps(internal=1)
904 p += pkt
905 lst.append((self, pl))
907 return p, lst
909 def build_ps(self, internal=0):
910 # type: (int) -> Tuple[bytes, List[Tuple[Packet, List[Tuple[Any, Any, bytes]]]]] # noqa: E501
911 p, lst = self.do_build_ps()
912# if not internal:
913# pkt = self
914# while pkt.haslayer(conf.padding_layer):
915# pkt = pkt.getlayer(conf.padding_layer)
916# lst.append( (pkt, [ ("loakjkjd", pkt.load, pkt.load) ] ) )
917# p += pkt.load
918# pkt = pkt.payload
919 return p, lst
921 def canvas_dump(self, layer_shift=0, rebuild=1):
922 # type: (int, int) -> pyx.canvas.canvas
923 if PYX == 0:
924 raise ImportError("PyX and its dependencies must be installed")
925 canvas = pyx.canvas.canvas()
926 if rebuild:
927 _, t = self.__class__(raw(self)).build_ps()
928 else:
929 _, t = self.build_ps()
930 YTXTI = len(t)
931 for _, l in t:
932 YTXTI += len(l)
933 YTXT = float(YTXTI)
934 YDUMP = YTXT
936 XSTART = 1
937 XDSTART = 10
938 y = 0.0
939 yd = 0.0
940 XMUL = 0.55
941 YMUL = 0.4
943 backcolor = colgen(0.6, 0.8, 1.0, trans=pyx.color.rgb)
944 forecolor = colgen(0.2, 0.5, 0.8, trans=pyx.color.rgb)
945# backcolor=makecol(0.376, 0.729, 0.525, 1.0)
947 def hexstr(x):
948 # type: (bytes) -> str
949 return " ".join("%02x" % c for c in x)
951 def make_dump_txt(x, y, txt):
952 # type: (int, float, bytes) -> pyx.text.text
953 return pyx.text.text(
954 XDSTART + x * XMUL,
955 (YDUMP - y) * YMUL,
956 r"\tt{%s}" % hexstr(txt),
957 [pyx.text.size.Large]
958 )
960 def make_box(o):
961 # type: (pyx.bbox.bbox) -> pyx.bbox.bbox
962 return pyx.box.rect(
963 o.left(), o.bottom(), o.width(), o.height(),
964 relcenter=(0.5, 0.5)
965 )
967 def make_frame(lst):
968 # type: (List[Any]) -> pyx.path.path
969 if len(lst) == 1:
970 b = lst[0].bbox()
971 b.enlarge(pyx.unit.u_pt)
972 return b.path()
973 else:
974 fb = lst[0].bbox()
975 fb.enlarge(pyx.unit.u_pt)
976 lb = lst[-1].bbox()
977 lb.enlarge(pyx.unit.u_pt)
978 if len(lst) == 2 and fb.left() > lb.right():
979 return pyx.path.path(pyx.path.moveto(fb.right(), fb.top()),
980 pyx.path.lineto(fb.left(), fb.top()),
981 pyx.path.lineto(fb.left(), fb.bottom()), # noqa: E501
982 pyx.path.lineto(fb.right(), fb.bottom()), # noqa: E501
983 pyx.path.moveto(lb.left(), lb.top()),
984 pyx.path.lineto(lb.right(), lb.top()),
985 pyx.path.lineto(lb.right(), lb.bottom()), # noqa: E501
986 pyx.path.lineto(lb.left(), lb.bottom())) # noqa: E501
987 else:
988 # XXX
989 gb = lst[1].bbox()
990 if gb != lb:
991 gb.enlarge(pyx.unit.u_pt)
992 kb = lst[-2].bbox()
993 if kb != gb and kb != lb:
994 kb.enlarge(pyx.unit.u_pt)
995 return pyx.path.path(pyx.path.moveto(fb.left(), fb.top()),
996 pyx.path.lineto(fb.right(), fb.top()),
997 pyx.path.lineto(fb.right(), kb.bottom()), # noqa: E501
998 pyx.path.lineto(lb.right(), kb.bottom()), # noqa: E501
999 pyx.path.lineto(lb.right(), lb.bottom()), # noqa: E501
1000 pyx.path.lineto(lb.left(), lb.bottom()), # noqa: E501
1001 pyx.path.lineto(lb.left(), gb.top()),
1002 pyx.path.lineto(fb.left(), gb.top()),
1003 pyx.path.closepath(),)
1005 def make_dump(s, # type: bytes
1006 shift=0, # type: int
1007 y=0., # type: float
1008 col=None, # type: pyx.color.color
1009 bkcol=None, # type: pyx.color.color
1010 large=16 # type: int
1011 ):
1012 # type: (...) -> Tuple[pyx.canvas.canvas, pyx.bbox.bbox, int, float] # noqa: E501
1013 c = pyx.canvas.canvas()
1014 tlist = []
1015 while s:
1016 dmp, s = s[:large - shift], s[large - shift:]
1017 txt = make_dump_txt(shift, y, dmp)
1018 tlist.append(txt)
1019 shift += len(dmp)
1020 if shift >= 16:
1021 shift = 0
1022 y += 1
1023 if col is None:
1024 col = pyx.color.rgb.red
1025 if bkcol is None:
1026 bkcol = pyx.color.rgb.white
1027 c.stroke(make_frame(tlist), [col, pyx.deco.filled([bkcol]), pyx.style.linewidth.Thick]) # noqa: E501
1028 for txt in tlist:
1029 c.insert(txt)
1030 return c, tlist[-1].bbox(), shift, y
1032 last_shift, last_y = 0, 0.0
1033 while t:
1034 bkcol = next(backcolor)
1035 proto, fields = t.pop()
1036 y += 0.5
1037 pt = pyx.text.text(
1038 XSTART,
1039 (YTXT - y) * YMUL,
1040 r"\font\cmssfont=cmss10\cmssfont{%s}" % tex_escape(
1041 str(proto.name)
1042 ),
1043 [pyx.text.size.Large]
1044 )
1045 y += 1
1046 ptbb = pt.bbox()
1047 ptbb.enlarge(pyx.unit.u_pt * 2)
1048 canvas.stroke(ptbb.path(), [pyx.color.rgb.black, pyx.deco.filled([bkcol])]) # noqa: E501
1049 canvas.insert(pt)
1050 for field, fval, fdump in fields:
1051 col = next(forecolor)
1052 ft = pyx.text.text(XSTART, (YTXT - y) * YMUL, r"\font\cmssfont=cmss10\cmssfont{%s}" % tex_escape(field.name)) # noqa: E501
1053 if isinstance(field, BitField):
1054 fsize = '%sb' % field.size
1055 else:
1056 fsize = '%sB' % len(fdump)
1057 if (hasattr(field, 'field') and
1058 'LE' in field.field.__class__.__name__[:3] or
1059 'LE' in field.__class__.__name__[:3]):
1060 fsize = r'$\scriptstyle\langle$' + fsize
1061 st = pyx.text.text(XSTART + 3.4, (YTXT - y) * YMUL, r"\font\cmbxfont=cmssbx10 scaled 600\cmbxfont{%s}" % fsize, [pyx.text.halign.boxright]) # noqa: E501
1062 if isinstance(fval, str):
1063 if len(fval) > 18:
1064 fval = fval[:18] + "[...]"
1065 else:
1066 fval = ""
1067 vt = pyx.text.text(XSTART + 3.5, (YTXT - y) * YMUL, r"\font\cmssfont=cmss10\cmssfont{%s}" % tex_escape(fval)) # noqa: E501
1068 y += 1.0
1069 if fdump:
1070 dt, target, last_shift, last_y = make_dump(fdump, last_shift, last_y, col, bkcol) # noqa: E501
1072 dtb = target
1073 vtb = vt.bbox()
1074 bxvt = make_box(vtb)
1075 bxdt = make_box(dtb)
1076 dtb.enlarge(pyx.unit.u_pt)
1077 try:
1078 if yd < 0:
1079 cnx = pyx.connector.curve(bxvt, bxdt, absangle1=0, absangle2=-90) # noqa: E501
1080 else:
1081 cnx = pyx.connector.curve(bxvt, bxdt, absangle1=0, absangle2=90) # noqa: E501
1082 except Exception:
1083 pass
1084 else:
1085 canvas.stroke(cnx, [pyx.style.linewidth.thin, pyx.deco.earrow.small, col]) # noqa: E501
1087 canvas.insert(dt)
1089 canvas.insert(ft)
1090 canvas.insert(st)
1091 canvas.insert(vt)
1092 last_y += layer_shift
1094 return canvas
1096 def extract_padding(self, s):
1097 # type: (bytes) -> Tuple[bytes, Optional[bytes]]
1098 """
1099 DEV: to be overloaded to extract current layer's padding.
1101 :param str s: the current layer
1102 :return: a couple of strings (actual layer, padding)
1103 """
1104 return s, None
1106 def post_dissect(self, s):
1107 # type: (bytes) -> bytes
1108 """DEV: is called right after the current layer has been dissected"""
1109 return s
1111 def pre_dissect(self, s):
1112 # type: (bytes) -> bytes
1113 """DEV: is called right before the current layer is dissected"""
1114 return s
1116 def do_dissect(self, s):
1117 # type: (bytes) -> bytes
1118 _raw = s
1119 self.raw_packet_cache_fields = {}
1120 for f in self.fields_desc:
1121 s, fval = f.getfield(self, s)
1122 # Skip unused ConditionalField
1123 if f.isconditional and fval is None:
1124 continue
1125 # We need to track fields with mutable values to discard
1126 # .raw_packet_cache when needed.
1127 if (f.islist or f.holds_packets or f.ismutable) and fval is not None:
1128 self.raw_packet_cache_fields[f.name] = \
1129 self._raw_packet_cache_field_value(f, fval, copy=True)
1130 self.fields[f.name] = fval
1131 # Nothing left to dissect
1132 if not s and (f.ismayend or
1133 (fval is not None and f.isconditional and
1134 f.fld.ismayend)): # type: ignore
1135 break
1136 self.raw_packet_cache = _raw[:-len(s)] if s else _raw
1137 self.explicit = 1
1138 return s
1140 def do_dissect_payload(self, s):
1141 # type: (bytes) -> None
1142 """
1143 Perform the dissection of the layer's payload
1145 :param str s: the raw layer
1146 """
1147 if s:
1148 if (
1149 self.stop_dissection_after and
1150 isinstance(self, self.stop_dissection_after)
1151 ):
1152 # stop dissection here
1153 p = conf.raw_layer(s, _internal=1, _underlayer=self)
1154 self.add_payload(p)
1155 return
1156 cls = self.guess_payload_class(s)
1157 try:
1158 p = cls(
1159 s,
1160 stop_dissection_after=self.stop_dissection_after,
1161 _internal=1,
1162 _underlayer=self,
1163 )
1164 except KeyboardInterrupt:
1165 raise
1166 except Exception:
1167 if conf.debug_dissector:
1168 if issubtype(cls, Packet):
1169 log_runtime.error("%s dissector failed", cls.__name__)
1170 else:
1171 log_runtime.error("%s.guess_payload_class() returned "
1172 "[%s]",
1173 self.__class__.__name__, repr(cls))
1174 if cls is not None:
1175 raise
1176 p = conf.raw_layer(s, _internal=1, _underlayer=self)
1177 self.add_payload(p)
1179 def dissect(self, s):
1180 # type: (bytes) -> None
1181 s = self.pre_dissect(s)
1183 s = self.do_dissect(s)
1185 s = self.post_dissect(s)
1187 payl, pad = self.extract_padding(s)
1188 self.do_dissect_payload(payl)
1189 if pad and conf.padding:
1190 self.add_payload(conf.padding_layer(pad))
1192 def guess_payload_class(self, payload):
1193 # type: (bytes) -> Type[Packet]
1194 """
1195 DEV: Guesses the next payload class from layer bonds.
1196 Can be overloaded to use a different mechanism.
1198 :param str payload: the layer's payload
1199 :return: the payload class
1200 """
1201 for t in self.aliastypes:
1202 for fval, cls in t.payload_guess:
1203 try:
1204 if all(v == self.getfieldval(k)
1205 for k, v in fval.items()):
1206 return cls # type: ignore
1207 except AttributeError:
1208 pass
1209 return self.default_payload_class(payload)
1211 def default_payload_class(self, payload):
1212 # type: (bytes) -> Type[Packet]
1213 """
1214 DEV: Returns the default payload class if nothing has been found by the
1215 guess_payload_class() method.
1217 :param str payload: the layer's payload
1218 :return: the default payload class define inside the configuration file
1219 """
1220 return conf.raw_layer
1222 def hide_defaults(self):
1223 # type: () -> None
1224 """Removes fields' values that are the same as default values."""
1225 # use list(): self.fields is modified in the loop
1226 for k, v in list(self.fields.items()):
1227 v = self.fields[k]
1228 if k in self.default_fields:
1229 if self.default_fields[k] == v:
1230 del self.fields[k]
1231 self.payload.hide_defaults()
1233 def clone_with(self, payload=None, **kargs):
1234 # type: (Optional[Any], **Any) -> Any
1235 pkt = self.__class__()
1236 pkt.explicit = 1
1237 pkt.fields = kargs
1238 pkt.default_fields = self.copy_fields_dict(self.default_fields)
1239 pkt.overloaded_fields = self.overloaded_fields.copy()
1240 pkt.time = self.time
1241 pkt.underlayer = self.underlayer
1242 pkt.parent = self.parent
1243 pkt.post_transforms = self.post_transforms
1244 pkt.raw_packet_cache = self.raw_packet_cache
1245 pkt.raw_packet_cache_fields = self.copy_fields_dict(
1246 self.raw_packet_cache_fields
1247 )
1248 pkt.wirelen = self.wirelen
1249 pkt.comments = self.comments
1250 pkt.sniffed_on = self.sniffed_on
1251 pkt.direction = self.direction
1252 if payload is not None:
1253 pkt.add_payload(payload)
1254 return pkt
1256 def __iter__(self):
1257 # type: () -> Iterator[Packet]
1258 """Iterates through all sub-packets generated by this Packet."""
1259 def loop(todo, done, self=self):
1260 # type: (List[str], Dict[str, Any], Any) -> Iterator[Packet]
1261 if todo:
1262 eltname = todo.pop()
1263 elt = self.getfieldval(eltname)
1264 if not isinstance(elt, Gen):
1265 if self.get_field(eltname).islist:
1266 elt = SetGen([elt])
1267 else:
1268 elt = SetGen(elt)
1269 for e in elt:
1270 done[eltname] = e
1271 for x in loop(todo[:], done):
1272 yield x
1273 else:
1274 if isinstance(self.payload, NoPayload):
1275 payloads = SetGen([None]) # type: SetGen[Packet]
1276 else:
1277 payloads = self.payload
1278 for payl in payloads:
1279 # Let's make sure subpackets are consistent
1280 done2 = done.copy()
1281 for k in done2:
1282 if isinstance(done2[k], VolatileValue):
1283 done2[k] = done2[k]._fix()
1284 pkt = self.clone_with(payload=payl, **done2)
1285 yield pkt
1287 if self.explicit or self.raw_packet_cache is not None:
1288 todo = []
1289 done = self.fields
1290 else:
1291 todo = [k for (k, v) in itertools.chain(self.default_fields.items(),
1292 self.overloaded_fields.items())
1293 if isinstance(v, VolatileValue)] + list(self.fields)
1294 done = {}
1295 return loop(todo, done)
1297 def iterpayloads(self):
1298 # type: () -> Iterator[Packet]
1299 """Used to iter through the payloads of a Packet.
1300 Useful for DNS or 802.11 for instance.
1301 """
1302 yield self
1303 current = self
1304 while current.payload:
1305 current = current.payload
1306 yield current
1308 def __gt__(self, other):
1309 # type: (Packet) -> int
1310 """True if other is an answer from self (self ==> other)."""
1311 if isinstance(other, Packet):
1312 return other < self
1313 elif isinstance(other, bytes):
1314 return 1
1315 else:
1316 raise TypeError((self, other))
1318 def __lt__(self, other):
1319 # type: (Packet) -> int
1320 """True if self is an answer from other (other ==> self)."""
1321 if isinstance(other, Packet):
1322 return self.answers(other)
1323 elif isinstance(other, bytes):
1324 return 1
1325 else:
1326 raise TypeError((self, other))
1328 def __eq__(self, other):
1329 # type: (Any) -> bool
1330 if not isinstance(other, self.__class__):
1331 return False
1332 for f in self.fields_desc:
1333 if f not in other.fields_desc:
1334 return False
1335 if self.getfieldval(f.name) != other.getfieldval(f.name):
1336 return False
1337 return self.payload == other.payload
1339 def __ne__(self, other):
1340 # type: (Any) -> bool
1341 return not self.__eq__(other)
1343 # Note: setting __hash__ to None is the standard way
1344 # of making an object un-hashable. mypy doesn't know that
1345 __hash__ = None # type: ignore
1347 def hashret(self):
1348 # type: () -> bytes
1349 """DEV: returns a string that has the same value for a request
1350 and its answer."""
1351 return self.payload.hashret()
1353 def answers(self, other):
1354 # type: (Packet) -> int
1355 """DEV: true if self is an answer from other"""
1356 if other.__class__ == self.__class__:
1357 return self.payload.answers(other.payload)
1358 return 0
1360 def layers(self):
1361 # type: () -> List[Type[Packet]]
1362 """returns a list of layer classes (including subclasses) in this packet""" # noqa: E501
1363 layers = []
1364 lyr = self # type: Optional[Packet]
1365 while lyr:
1366 layers.append(lyr.__class__)
1367 lyr = lyr.payload.getlayer(0, _subclass=True)
1368 return layers
1370 def haslayer(self, cls, _subclass=None):
1371 # type: (Union[Type[Packet], str], Optional[bool]) -> int
1372 """
1373 true if self has a layer that is an instance of cls.
1374 Superseded by "cls in self" syntax.
1375 """
1376 if _subclass is None:
1377 _subclass = self.match_subclass or None
1378 if _subclass:
1379 match = issubtype
1380 else:
1381 match = lambda x, t: bool(x == t)
1382 if cls is None or match(self.__class__, cls) \
1383 or cls in [self.__class__.__name__, self._name]:
1384 return True
1385 for f in self.packetfields:
1386 fvalue_gen = self.getfieldval(f.name)
1387 if fvalue_gen is None:
1388 continue
1389 if not f.islist:
1390 fvalue_gen = SetGen(fvalue_gen, _iterpacket=0)
1391 for fvalue in fvalue_gen:
1392 if isinstance(fvalue, Packet):
1393 ret = fvalue.haslayer(cls, _subclass=_subclass)
1394 if ret:
1395 return ret
1396 return self.payload.haslayer(cls, _subclass=_subclass)
1398 def getlayer(self,
1399 cls, # type: Union[int, Type[Packet], str]
1400 nb=1, # type: int
1401 _track=None, # type: Optional[List[int]]
1402 _subclass=None, # type: Optional[bool]
1403 **flt # type: Any
1404 ):
1405 # type: (...) -> Optional[Packet]
1406 """Return the nb^th layer that is an instance of cls, matching flt
1407values.
1408 """
1409 if _subclass is None:
1410 _subclass = self.match_subclass or None
1411 if _subclass:
1412 match = issubtype
1413 else:
1414 match = lambda x, t: bool(x == t)
1415 # Note:
1416 # cls can be int, packet, str
1417 # string_class_name can be packet, str (packet or packet+field)
1418 # class_name can be packet, str (packet only)
1419 if isinstance(cls, int):
1420 nb = cls + 1
1421 string_class_name = "" # type: Union[Type[Packet], str]
1422 else:
1423 string_class_name = cls
1424 class_name = "" # type: Union[Type[Packet], str]
1425 fld = None # type: Optional[str]
1426 if isinstance(string_class_name, str) and "." in string_class_name:
1427 class_name, fld = string_class_name.split(".", 1)
1428 else:
1429 class_name, fld = string_class_name, None
1430 if not class_name or match(self.__class__, class_name) \
1431 or class_name in [self.__class__.__name__, self._name]:
1432 if all(self.getfieldval(fldname) == fldvalue
1433 for fldname, fldvalue in flt.items()):
1434 if nb == 1:
1435 if fld is None:
1436 return self
1437 else:
1438 return self.getfieldval(fld) # type: ignore
1439 else:
1440 nb -= 1
1441 for f in self.packetfields:
1442 fvalue_gen = self.getfieldval(f.name)
1443 if fvalue_gen is None:
1444 continue
1445 if not f.islist:
1446 fvalue_gen = SetGen(fvalue_gen, _iterpacket=0)
1447 for fvalue in fvalue_gen:
1448 if isinstance(fvalue, Packet):
1449 track = [] # type: List[int]
1450 ret = fvalue.getlayer(class_name, nb=nb, _track=track,
1451 _subclass=_subclass, **flt)
1452 if ret is not None:
1453 return ret
1454 nb = track[0]
1455 return self.payload.getlayer(class_name, nb=nb, _track=_track,
1456 _subclass=_subclass, **flt)
1458 def firstlayer(self):
1459 # type: () -> Packet
1460 q = self
1461 while q.underlayer is not None:
1462 q = q.underlayer
1463 return q
1465 def __getitem__(self, cls):
1466 # type: (Union[Type[Packet], str]) -> Any
1467 if isinstance(cls, slice):
1468 lname = cls.start
1469 if cls.stop:
1470 ret = self.getlayer(cls.start, nb=cls.stop, **(cls.step or {}))
1471 else:
1472 ret = self.getlayer(cls.start, **(cls.step or {}))
1473 else:
1474 lname = cls
1475 ret = self.getlayer(cls)
1476 if ret is None:
1477 if isinstance(lname, type):
1478 name = lname.__name__
1479 elif not isinstance(lname, bytes):
1480 name = repr(lname)
1481 else:
1482 name = cast(str, lname)
1483 raise IndexError("Layer [%s] not found" % name)
1484 return ret
1486 def __delitem__(self, cls):
1487 # type: (Type[Packet]) -> None
1488 del self[cls].underlayer.payload
1490 def __setitem__(self, cls, val):
1491 # type: (Type[Packet], Packet) -> None
1492 self[cls].underlayer.payload = val
1494 def __contains__(self, cls):
1495 # type: (Union[Type[Packet], str]) -> int
1496 """
1497 "cls in self" returns true if self has a layer which is an
1498 instance of cls.
1499 """
1500 return self.haslayer(cls)
1502 def route(self):
1503 # type: () -> Tuple[Optional[str], Optional[str], Optional[str]]
1504 return self.payload.route()
1506 def fragment(self, *args, **kargs):
1507 # type: (*Any, **Any) -> List[Packet]
1508 return self.payload.fragment(*args, **kargs)
1510 def display(self, *args, **kargs): # Deprecated. Use show()
1511 # type: (*Any, **Any) -> None
1512 """Deprecated. Use show() method."""
1513 self.show(*args, **kargs)
1515 def _show_or_dump(self,
1516 dump=False, # type: bool
1517 indent=3, # type: int
1518 lvl="", # type: str
1519 label_lvl="", # type: str
1520 first_call=True # type: bool
1521 ):
1522 # type: (...) -> Optional[str]
1523 """
1524 Internal method that shows or dumps a hierarchical view of a packet.
1525 Called by show.
1527 :param dump: determine if it prints or returns the string value
1528 :param int indent: the size of indentation for each layer
1529 :param str lvl: additional information about the layer lvl
1530 :param str label_lvl: additional information about the layer fields
1531 :param first_call: determine if the current function is the first
1532 :return: return a hierarchical view if dump, else print it
1533 """
1535 if dump:
1536 from scapy.themes import ColorTheme, AnsiColorTheme
1537 ct: ColorTheme = AnsiColorTheme() # No color for dump output
1538 else:
1539 ct = conf.color_theme
1540 s = "%s%s %s %s\n" % (label_lvl,
1541 ct.punct("###["),
1542 ct.layer_name(self.name),
1543 ct.punct("]###"))
1544 fields = self.fields_desc.copy()
1545 while fields:
1546 f = fields.pop(0)
1547 if isinstance(f, ConditionalField) and not f._evalcond(self):
1548 continue
1549 if hasattr(f, "fields"): # Field has subfields
1550 s += "%s %s =\n" % (
1551 label_lvl + lvl,
1552 ct.depreciate_field_name(f.name),
1553 )
1554 lvl += " " * indent * self.show_indent
1555 for i, fld in enumerate(x for x in f.fields if hasattr(self, x.name)):
1556 fields.insert(i, fld)
1557 continue
1558 if isinstance(f, Emph) or f in conf.emph:
1559 ncol = ct.emph_field_name
1560 vcol = ct.emph_field_value
1561 else:
1562 ncol = ct.field_name
1563 vcol = ct.field_value
1564 pad = max(0, 10 - len(f.name)) * " "
1565 fvalue = self.getfieldval(f.name)
1566 if isinstance(fvalue, Packet) or (f.islist and f.holds_packets and isinstance(fvalue, list)): # noqa: E501
1567 s += "%s %s%s%s%s\n" % (label_lvl + lvl,
1568 ct.punct("\\"),
1569 ncol(f.name),
1570 pad,
1571 ct.punct("\\"))
1572 fvalue_gen = SetGen(
1573 fvalue,
1574 _iterpacket=0
1575 ) # type: SetGen[Packet]
1576 for fvalue in fvalue_gen:
1577 s += fvalue._show_or_dump(dump=dump, indent=indent, label_lvl=label_lvl + lvl + " |", first_call=False) # noqa: E501
1578 else:
1579 begn = "%s %s%s%s " % (label_lvl + lvl,
1580 ncol(f.name),
1581 pad,
1582 ct.punct("="),)
1583 reprval = f.i2repr(self, fvalue)
1584 if isinstance(reprval, str):
1585 reprval = reprval.replace("\n", "\n" + " " * (len(label_lvl) + # noqa: E501
1586 len(lvl) +
1587 len(f.name) +
1588 4))
1589 s += "%s%s\n" % (begn, vcol(reprval))
1590 if self.payload:
1591 s += self.payload._show_or_dump( # type: ignore
1592 dump=dump,
1593 indent=indent,
1594 lvl=lvl + (" " * indent * self.show_indent),
1595 label_lvl=label_lvl,
1596 first_call=False
1597 )
1599 if first_call and not dump:
1600 print(s)
1601 return None
1602 else:
1603 return s
1605 def show(self, dump=False, indent=3, lvl="", label_lvl=""):
1606 # type: (bool, int, str, str) -> Optional[Any]
1607 """
1608 Prints or returns (when "dump" is true) a hierarchical view of the
1609 packet.
1611 :param dump: determine if it prints or returns the string value
1612 :param int indent: the size of indentation for each layer
1613 :param str lvl: additional information about the layer lvl
1614 :param str label_lvl: additional information about the layer fields
1615 :return: return a hierarchical view if dump, else print it
1616 """
1617 return self._show_or_dump(dump, indent, lvl, label_lvl)
1619 def show2(self, dump=False, indent=3, lvl="", label_lvl=""):
1620 # type: (bool, int, str, str) -> Optional[Any]
1621 """
1622 Prints or returns (when "dump" is true) a hierarchical view of an
1623 assembled version of the packet, so that automatic fields are
1624 calculated (checksums, etc.)
1626 :param dump: determine if it prints or returns the string value
1627 :param int indent: the size of indentation for each layer
1628 :param str lvl: additional information about the layer lvl
1629 :param str label_lvl: additional information about the layer fields
1630 :return: return a hierarchical view if dump, else print it
1631 """
1632 return self.__class__(raw(self)).show(dump, indent, lvl, label_lvl)
1634 def sprintf(self, fmt, relax=1):
1635 # type: (str, int) -> str
1636 """
1637 sprintf(format, [relax=1]) -> str
1639 Where format is a string that can include directives. A directive
1640 begins and ends by % and has the following format:
1641 ``%[fmt[r],][cls[:nb].]field%``
1643 :param fmt: is a classic printf directive, "r" can be appended for raw
1644 substitution:
1645 (ex: IP.flags=0x18 instead of SA), nb is the number of the layer
1646 (ex: for IP/IP packets, IP:2.src is the src of the upper IP layer).
1647 Special case : "%.time%" is the creation time.
1648 Ex::
1650 p.sprintf(
1651 "%.time% %-15s,IP.src% -> %-15s,IP.dst% %IP.chksum% "
1652 "%03xr,IP.proto% %r,TCP.flags%"
1653 )
1655 Moreover, the format string can include conditional statements. A
1656 conditional statement looks like : {layer:string} where layer is a
1657 layer name, and string is the string to insert in place of the
1658 condition if it is true, i.e. if layer is present. If layer is
1659 preceded by a "!", the result is inverted. Conditions can be
1660 imbricated. A valid statement can be::
1662 p.sprintf("This is a{TCP: TCP}{UDP: UDP}{ICMP:n ICMP} packet")
1663 p.sprintf("{IP:%IP.dst% {ICMP:%ICMP.type%}{TCP:%TCP.dport%}}")
1665 A side effect is that, to obtain "{" and "}" characters, you must use
1666 "%(" and "%)".
1667 """
1669 escape = {"%": "%",
1670 "(": "{",
1671 ")": "}"}
1673 # Evaluate conditions
1674 while "{" in fmt:
1675 i = fmt.rindex("{")
1676 j = fmt[i + 1:].index("}")
1677 cond = fmt[i + 1:i + j + 1]
1678 k = cond.find(":")
1679 if k < 0:
1680 raise Scapy_Exception("Bad condition in format string: [%s] (read sprintf doc!)" % cond) # noqa: E501
1681 cond, format_ = cond[:k], cond[k + 1:]
1682 res = False
1683 if cond[0] == "!":
1684 res = True
1685 cond = cond[1:]
1686 if self.haslayer(cond):
1687 res = not res
1688 if not res:
1689 format_ = ""
1690 fmt = fmt[:i] + format_ + fmt[i + j + 2:]
1692 # Evaluate directives
1693 s = ""
1694 while "%" in fmt:
1695 i = fmt.index("%")
1696 s += fmt[:i]
1697 fmt = fmt[i + 1:]
1698 if fmt and fmt[0] in escape:
1699 s += escape[fmt[0]]
1700 fmt = fmt[1:]
1701 continue
1702 try:
1703 i = fmt.index("%")
1704 sfclsfld = fmt[:i]
1705 fclsfld = sfclsfld.split(",")
1706 if len(fclsfld) == 1:
1707 f = "s"
1708 clsfld = fclsfld[0]
1709 elif len(fclsfld) == 2:
1710 f, clsfld = fclsfld
1711 else:
1712 raise Scapy_Exception
1713 if "." in clsfld:
1714 cls, fld = clsfld.split(".")
1715 else:
1716 cls = self.__class__.__name__
1717 fld = clsfld
1718 num = 1
1719 if ":" in cls:
1720 cls, snum = cls.split(":")
1721 num = int(snum)
1722 fmt = fmt[i + 1:]
1723 except Exception:
1724 raise Scapy_Exception("Bad format string [%%%s%s]" % (fmt[:25], fmt[25:] and "...")) # noqa: E501
1725 else:
1726 if fld == "time":
1727 val = time.strftime(
1728 "%H:%M:%S.%%06i",
1729 time.localtime(float(self.time))
1730 ) % int((self.time - int(self.time)) * 1000000)
1731 elif cls == self.__class__.__name__ and hasattr(self, fld):
1732 if num > 1:
1733 val = self.payload.sprintf("%%%s,%s:%s.%s%%" % (f, cls, num - 1, fld), relax) # noqa: E501
1734 f = "s"
1735 else:
1736 try:
1737 val = self.getfieldval(fld)
1738 except AttributeError:
1739 val = getattr(self, fld)
1740 if f[-1] == "r": # Raw field value
1741 f = f[:-1]
1742 if not f:
1743 f = "s"
1744 else:
1745 if fld in self.fieldtype:
1746 val = self.fieldtype[fld].i2repr(self, val)
1747 else:
1748 val = self.payload.sprintf("%%%s%%" % sfclsfld, relax)
1749 f = "s"
1750 s += ("%" + f) % val
1752 s += fmt
1753 return s
1755 def mysummary(self):
1756 # type: () -> str
1757 """DEV: can be overloaded to return a string that summarizes the layer.
1758 Only one mysummary() is used in a whole packet summary: the one of the upper layer, # noqa: E501
1759 except if a mysummary() also returns (as a couple) a list of layers whose # noqa: E501
1760 mysummary() must be called if they are present."""
1761 return ""
1763 def _do_summary(self):
1764 # type: () -> Tuple[int, str, List[Any]]
1765 found, s, needed = self.payload._do_summary()
1766 ret = ""
1767 if not found or self.__class__ in needed:
1768 ret = self.mysummary()
1769 if isinstance(ret, tuple):
1770 ret, n = ret
1771 needed += n
1772 if ret or needed:
1773 found = 1
1774 if not ret:
1775 ret = self.__class__.__name__ if self.show_summary else ""
1776 if self.__class__ in conf.emph:
1777 impf = []
1778 for f in self.fields_desc:
1779 if f in conf.emph:
1780 impf.append("%s=%s" % (f.name, f.i2repr(self, self.getfieldval(f.name)))) # noqa: E501
1781 ret = "%s [%s]" % (ret, " ".join(impf))
1782 if ret and s:
1783 ret = "%s / %s" % (ret, s)
1784 else:
1785 ret = "%s%s" % (ret, s)
1786 return found, ret, needed
1788 def summary(self, intern=0):
1789 # type: (int) -> str
1790 """Prints a one line summary of a packet."""
1791 return self._do_summary()[1]
1793 def lastlayer(self, layer=None):
1794 # type: (Optional[Packet]) -> Packet
1795 """Returns the uppest layer of the packet"""
1796 return self.payload.lastlayer(self)
1798 def decode_payload_as(self, cls):
1799 # type: (Type[Packet]) -> None
1800 """Reassembles the payload and decode it using another packet class"""
1801 s = raw(self.payload)
1802 self.payload = cls(s, _internal=1, _underlayer=self)
1803 pp = self
1804 while pp.underlayer is not None:
1805 pp = pp.underlayer
1806 self.payload.dissection_done(pp)
1808 def _command(self, json=False):
1809 # type: (bool) -> List[Tuple[str, Any]]
1810 """
1811 Internal method used to generate command() and json()
1812 """
1813 f = []
1814 iterator: Iterator[Tuple[str, Any]]
1815 if json:
1816 iterator = ((x.name, self.getfieldval(x.name)) for x in self.fields_desc)
1817 else:
1818 iterator = iter(self.fields.items())
1819 for fn, fv in iterator:
1820 fld = self.get_field(fn)
1821 if isinstance(fv, (list, dict, set)) and not fv and not fld.default:
1822 continue
1823 if isinstance(fv, Packet):
1824 if json:
1825 fv = {k: v for (k, v) in fv._command(json=True)}
1826 else:
1827 fv = fv.command()
1828 elif fld.islist and fld.holds_packets and isinstance(fv, list):
1829 if json:
1830 fv = [
1831 {k: v for (k, v) in x}
1832 for x in map(lambda y: Packet._command(y, json=True), fv)
1833 ]
1834 else:
1835 fv = "[%s]" % ",".join(map(Packet.command, fv))
1836 elif fld.islist and isinstance(fv, list):
1837 if json:
1838 fv = [
1839 getattr(x, 'command', lambda: repr(x))()
1840 for x in fv
1841 ]
1842 else:
1843 fv = "[%s]" % ",".join(
1844 getattr(x, 'command', lambda: repr(x))()
1845 for x in fv
1846 )
1847 elif isinstance(fv, FlagValue):
1848 fv = int(fv)
1849 elif callable(getattr(fv, 'command', None)):
1850 fv = fv.command(json=json)
1851 else:
1852 if json:
1853 if isinstance(fv, bytes):
1854 fv = fv.decode("utf-8", errors="backslashreplace")
1855 else:
1856 fv = fld.i2h(self, fv)
1857 else:
1858 fv = repr(fld.i2h(self, fv))
1859 f.append((fn, fv))
1860 return f
1862 def command(self):
1863 # type: () -> str
1864 """
1865 Returns a string representing the command you have to type to
1866 obtain the same packet
1867 """
1868 c = "%s(%s)" % (
1869 self.__class__.__name__,
1870 ", ".join("%s=%s" % x for x in self._command())
1871 )
1872 pc = self.payload.command()
1873 if pc:
1874 c += "/" + pc
1875 return c
1877 def json(self):
1878 # type: () -> str
1879 """
1880 Returns a JSON representing the packet.
1882 Please note that this cannot be used for bijective usage: data loss WILL occur,
1883 so it will not make sense to try to rebuild the packet from the output.
1884 This must only be used for a grepping/displaying purpose.
1885 """
1886 dump = json.dumps({k: v for (k, v) in self._command(json=True)})
1887 pc = self.payload.json()
1888 if pc:
1889 dump = dump[:-1] + ", \"payload\": %s}" % pc
1890 return dump
1893class NoPayload(Packet):
1894 def __new__(cls, *args, **kargs):
1895 # type: (Type[Packet], *Any, **Any) -> NoPayload
1896 singl = cls.__dict__.get("__singl__")
1897 if singl is None:
1898 cls.__singl__ = singl = Packet.__new__(cls)
1899 Packet.__init__(singl)
1900 return singl # type: ignore
1902 def __init__(self, *args, **kargs):
1903 # type: (*Any, **Any) -> None
1904 pass
1906 def dissection_done(self, pkt):
1907 # type: (Packet) -> None
1908 pass
1910 def add_payload(self, payload):
1911 # type: (Union[Packet, bytes]) -> NoReturn
1912 raise Scapy_Exception("Can't add payload to NoPayload instance")
1914 def remove_payload(self):
1915 # type: () -> None
1916 pass
1918 def add_underlayer(self, underlayer):
1919 # type: (Any) -> None
1920 pass
1922 def remove_underlayer(self, other):
1923 # type: (Packet) -> None
1924 pass
1926 def add_parent(self, parent):
1927 # type: (Any) -> None
1928 pass
1930 def remove_parent(self, other):
1931 # type: (Packet) -> None
1932 pass
1934 def copy(self):
1935 # type: () -> NoPayload
1936 return self
1938 def clear_cache(self):
1939 # type: () -> None
1940 pass
1942 def __repr__(self):
1943 # type: () -> str
1944 return ""
1946 def __str__(self):
1947 # type: () -> str
1948 return ""
1950 def __bytes__(self):
1951 # type: () -> bytes
1952 return b""
1954 def __nonzero__(self):
1955 # type: () -> bool
1956 return False
1957 __bool__ = __nonzero__
1959 def do_build(self):
1960 # type: () -> bytes
1961 return b""
1963 def build(self):
1964 # type: () -> bytes
1965 return b""
1967 def build_padding(self):
1968 # type: () -> bytes
1969 return b""
1971 def build_done(self, p):
1972 # type: (bytes) -> bytes
1973 return p
1975 def build_ps(self, internal=0):
1976 # type: (int) -> Tuple[bytes, List[Any]]
1977 return b"", []
1979 def getfieldval(self, attr):
1980 # type: (str) -> NoReturn
1981 raise AttributeError(attr)
1983 def getfield_and_val(self, attr):
1984 # type: (str) -> NoReturn
1985 raise AttributeError(attr)
1987 def setfieldval(self, attr, val):
1988 # type: (str, Any) -> NoReturn
1989 raise AttributeError(attr)
1991 def delfieldval(self, attr):
1992 # type: (str) -> NoReturn
1993 raise AttributeError(attr)
1995 def hide_defaults(self):
1996 # type: () -> None
1997 pass
1999 def __iter__(self):
2000 # type: () -> Iterator[Packet]
2001 return iter([])
2003 def __eq__(self, other):
2004 # type: (Any) -> bool
2005 if isinstance(other, NoPayload):
2006 return True
2007 return False
2009 def hashret(self):
2010 # type: () -> bytes
2011 return b""
2013 def answers(self, other):
2014 # type: (Packet) -> bool
2015 return isinstance(other, (NoPayload, conf.padding_layer)) # noqa: E501
2017 def haslayer(self, cls, _subclass=None):
2018 # type: (Union[Type[Packet], str], Optional[bool]) -> int
2019 return 0
2021 def getlayer(self,
2022 cls, # type: Union[int, Type[Packet], str]
2023 nb=1, # type: int
2024 _track=None, # type: Optional[List[int]]
2025 _subclass=None, # type: Optional[bool]
2026 **flt # type: Any
2027 ):
2028 # type: (...) -> Optional[Packet]
2029 if _track is not None:
2030 _track.append(nb)
2031 return None
2033 def fragment(self, *args, **kargs):
2034 # type: (*Any, **Any) -> List[Packet]
2035 raise Scapy_Exception("cannot fragment this packet")
2037 def show(self, dump=False, indent=3, lvl="", label_lvl=""):
2038 # type: (bool, int, str, str) -> None
2039 pass
2041 def sprintf(self, fmt, relax=1):
2042 # type: (str, int) -> str
2043 if relax:
2044 return "??"
2045 else:
2046 raise Scapy_Exception("Format not found [%s]" % fmt)
2048 def _do_summary(self):
2049 # type: () -> Tuple[int, str, List[Any]]
2050 return 0, "", []
2052 def layers(self):
2053 # type: () -> List[Type[Packet]]
2054 return []
2056 def lastlayer(self, layer=None):
2057 # type: (Optional[Packet]) -> Packet
2058 return layer or self
2060 def command(self):
2061 # type: () -> str
2062 return ""
2064 def json(self):
2065 # type: () -> str
2066 return ""
2068 def route(self):
2069 # type: () -> Tuple[None, None, None]
2070 return (None, None, None)
2073####################
2074# packet classes #
2075####################
2078class Raw(Packet):
2079 name = "Raw"
2080 fields_desc = [StrField("load", b"")]
2082 def __init__(self, _pkt=b"", *args, **kwargs):
2083 # type: (bytes, *Any, **Any) -> None
2084 if _pkt and not isinstance(_pkt, bytes):
2085 if isinstance(_pkt, tuple):
2086 _pkt, bn = _pkt
2087 _pkt = bytes_encode(_pkt), bn
2088 else:
2089 _pkt = bytes_encode(_pkt)
2090 super(Raw, self).__init__(_pkt, *args, **kwargs)
2092 def answers(self, other):
2093 # type: (Packet) -> int
2094 return 1
2096 def mysummary(self):
2097 # type: () -> str
2098 cs = conf.raw_summary
2099 if cs:
2100 if callable(cs):
2101 return "Raw %s" % cs(self.load)
2102 else:
2103 return "Raw %r" % self.load
2104 return Packet.mysummary(self)
2107class Padding(Raw):
2108 name = "Padding"
2110 def self_build(self):
2111 # type: (Optional[Any]) -> bytes
2112 return b""
2114 def build_padding(self):
2115 # type: () -> bytes
2116 return (
2117 bytes_encode(self.load) if self.raw_packet_cache is None
2118 else self.raw_packet_cache
2119 ) + self.payload.build_padding()
2122conf.raw_layer = Raw
2123conf.padding_layer = Padding
2124if conf.default_l2 is None:
2125 conf.default_l2 = Raw
2127#################
2128# Bind layers #
2129#################
2132def bind_bottom_up(lower, # type: Type[Packet]
2133 upper, # type: Type[Packet]
2134 __fval=None, # type: Optional[Any]
2135 **fval # type: Any
2136 ):
2137 # type: (...) -> None
2138 r"""Bind 2 layers for dissection.
2139 The upper layer will be chosen for dissection on top of the lower layer, if
2140 ALL the passed arguments are validated. If multiple calls are made with
2141 the same layers, the last one will be used as default.
2143 ex:
2144 >>> bind_bottom_up(Ether, SNAP, type=0x1234)
2145 >>> Ether(b'\xff\xff\xff\xff\xff\xff\xd0P\x99V\xdd\xf9\x124\x00\x00\x00\x00\x00') # noqa: E501
2146 <Ether dst=ff:ff:ff:ff:ff:ff src=d0:50:99:56:dd:f9 type=0x1234 |<SNAP OUI=0x0 code=0x0 |>> # noqa: E501
2147 """
2148 if __fval is not None:
2149 fval.update(__fval)
2150 lower.payload_guess = lower.payload_guess[:]
2151 lower.payload_guess.append((fval, upper))
2154def bind_top_down(lower, # type: Type[Packet]
2155 upper, # type: Type[Packet]
2156 __fval=None, # type: Optional[Any]
2157 **fval # type: Any
2158 ):
2159 # type: (...) -> None
2160 """Bind 2 layers for building.
2161 When the upper layer is added as a payload of the lower layer, all the
2162 arguments will be applied to them.
2164 ex:
2165 >>> bind_top_down(Ether, SNAP, type=0x1234)
2166 >>> Ether()/SNAP()
2167 <Ether type=0x1234 |<SNAP |>>
2168 """
2169 if __fval is not None:
2170 fval.update(__fval)
2171 upper._overload_fields = upper._overload_fields.copy() # type: ignore
2172 upper._overload_fields[lower] = fval
2175@conf.commands.register
2176def bind_layers(lower, # type: Type[Packet]
2177 upper, # type: Type[Packet]
2178 __fval=None, # type: Optional[Dict[str, int]]
2179 **fval # type: Any
2180 ):
2181 # type: (...) -> None
2182 """Bind 2 layers on some specific fields' values.
2184 It makes the packet being built and dissected when the arguments
2185 are present.
2187 This function calls both bind_bottom_up and bind_top_down, with
2188 all passed arguments.
2190 Please have a look at their docs:
2191 - help(bind_bottom_up)
2192 - help(bind_top_down)
2193 """
2194 if __fval is not None:
2195 fval.update(__fval)
2196 bind_top_down(lower, upper, **fval)
2197 bind_bottom_up(lower, upper, **fval)
2200def split_bottom_up(lower, # type: Type[Packet]
2201 upper, # type: Type[Packet]
2202 __fval=None, # type: Optional[Any]
2203 **fval # type: Any
2204 ):
2205 # type: (...) -> None
2206 """This call un-links an association that was made using bind_bottom_up.
2207 Have a look at help(bind_bottom_up)
2208 """
2209 if __fval is not None:
2210 fval.update(__fval)
2212 def do_filter(params, cls):
2213 # type: (Dict[str, int], Type[Packet]) -> bool
2214 params_is_invalid = any(
2215 k not in params or params[k] != v for k, v in fval.items()
2216 )
2217 return cls != upper or params_is_invalid
2218 lower.payload_guess = [x for x in lower.payload_guess if do_filter(*x)]
2221def split_top_down(lower, # type: Type[Packet]
2222 upper, # type: Type[Packet]
2223 __fval=None, # type: Optional[Any]
2224 **fval # type: Any
2225 ):
2226 # type: (...) -> None
2227 """This call un-links an association that was made using bind_top_down.
2228 Have a look at help(bind_top_down)
2229 """
2230 if __fval is not None:
2231 fval.update(__fval)
2232 if lower in upper._overload_fields:
2233 ofval = upper._overload_fields[lower]
2234 if any(k not in ofval or ofval[k] != v for k, v in fval.items()):
2235 return
2236 upper._overload_fields = upper._overload_fields.copy() # type: ignore
2237 del upper._overload_fields[lower]
2240@conf.commands.register
2241def split_layers(lower, # type: Type[Packet]
2242 upper, # type: Type[Packet]
2243 __fval=None, # type: Optional[Any]
2244 **fval # type: Any
2245 ):
2246 # type: (...) -> None
2247 """Split 2 layers previously bound.
2248 This call un-links calls bind_top_down and bind_bottom_up. It is the opposite of # noqa: E501
2249 bind_layers.
2251 Please have a look at their docs:
2252 - help(split_bottom_up)
2253 - help(split_top_down)
2254 """
2255 if __fval is not None:
2256 fval.update(__fval)
2257 split_bottom_up(lower, upper, **fval)
2258 split_top_down(lower, upper, **fval)
2261@conf.commands.register
2262def explore(layer=None):
2263 # type: (Optional[str]) -> None
2264 """Function used to discover the Scapy layers and protocols.
2265 It helps to see which packets exists in contrib or layer files.
2267 params:
2268 - layer: If specified, the function will explore the layer. If not,
2269 the GUI mode will be activated, to browse the available layers
2271 examples:
2272 >>> explore() # Launches the GUI
2273 >>> explore("dns") # Explore scapy.layers.dns
2274 >>> explore("http2") # Explore scapy.contrib.http2
2275 >>> explore(scapy.layers.bluetooth4LE)
2277 Note: to search a packet by name, use ls("name") rather than explore.
2278 """
2279 if layer is None: # GUI MODE
2280 if not conf.interactive:
2281 raise Scapy_Exception("explore() GUI-mode cannot be run in "
2282 "interactive mode. Please provide a "
2283 "'layer' parameter !")
2284 # 0 - Imports
2285 try:
2286 import prompt_toolkit
2287 except ImportError:
2288 raise ImportError("prompt_toolkit is not installed ! "
2289 "You may install IPython, which contains it, via"
2290 " `pip install ipython`")
2291 if not _version_checker(prompt_toolkit, (2, 0)):
2292 raise ImportError("prompt_toolkit >= 2.0.0 is required !")
2293 # Only available with prompt_toolkit > 2.0, not released on PyPi yet
2294 from prompt_toolkit.shortcuts.dialogs import radiolist_dialog, \
2295 button_dialog
2296 from prompt_toolkit.formatted_text import HTML
2297 # Check for prompt_toolkit >= 3.0.0
2298 call_ptk = lambda x: cast(str, x) # type: Callable[[Any], str]
2299 if _version_checker(prompt_toolkit, (3, 0)):
2300 call_ptk = lambda x: x.run()
2301 # 1 - Ask for layer or contrib
2302 btn_diag = button_dialog(
2303 title="Scapy v%s" % conf.version,
2304 text=HTML(
2305 '<style bg="white" fg="red">Chose the type of packets'
2306 ' you want to explore:</style>'
2307 ),
2308 buttons=[
2309 ("Layers", "layers"),
2310 ("Contribs", "contribs"),
2311 ("Cancel", "cancel")
2312 ])
2313 action = call_ptk(btn_diag)
2314 # 2 - Retrieve list of Packets
2315 if action == "layers":
2316 # Get all loaded layers
2317 lvalues = conf.layers.layers()
2318 # Restrict to layers-only (not contribs) + packet.py and asn1*.py
2319 values = [x for x in lvalues if ("layers" in x[0] or
2320 "packet" in x[0] or
2321 "asn1" in x[0])]
2322 elif action == "contribs":
2323 # Get all existing contribs
2324 from scapy.main import list_contrib
2325 cvalues = cast(List[Dict[str, str]], list_contrib(ret=True))
2326 values = [(x['name'], x['description'])
2327 for x in cvalues]
2328 # Remove very specific modules
2329 values = [x for x in values if "can" not in x[0]]
2330 else:
2331 # Escape/Cancel was pressed
2332 return
2333 # Build tree
2334 if action == "contribs":
2335 # A tree is a dictionary. Each layer contains a keyword
2336 # _l which contains the files in the layer, and a _name
2337 # argument which is its name. The other keys are the subfolders,
2338 # which are similar dictionaries
2339 tree = defaultdict(list) # type: Dict[str, Union[List[Any], Dict[str, Any]]] # noqa: E501
2340 for name, desc in values:
2341 if "." in name: # Folder detected
2342 parts = name.split(".")
2343 subtree = tree
2344 for pa in parts[:-1]:
2345 if pa not in subtree:
2346 subtree[pa] = {}
2347 # one layer deeper
2348 subtree = subtree[pa] # type: ignore
2349 subtree["_name"] = pa # type: ignore
2350 if "_l" not in subtree:
2351 subtree["_l"] = []
2352 subtree["_l"].append((parts[-1], desc)) # type: ignore
2353 else:
2354 tree["_l"].append((name, desc)) # type: ignore
2355 elif action == "layers":
2356 tree = {"_l": values}
2357 # 3 - Ask for the layer/contrib module to explore
2358 current = tree # type: Any
2359 previous = [] # type: List[Dict[str, Union[List[Any], Dict[str, Any]]]] # noqa: E501
2360 while True:
2361 # Generate tests & form
2362 folders = list(current.keys())
2363 _radio_values = [
2364 ("$" + name, str('[+] ' + name.capitalize()))
2365 for name in folders if not name.startswith("_")
2366 ] + current.get("_l", []) # type: List[str]
2367 cur_path = ""
2368 if previous:
2369 cur_path = ".".join(
2370 itertools.chain(
2371 (x["_name"] for x in previous[1:]), # type: ignore
2372 (current["_name"],)
2373 )
2374 )
2375 extra_text = (
2376 '\n<style bg="white" fg="green">> scapy.%s</style>'
2377 ) % (action + ("." + cur_path if cur_path else ""))
2378 # Show popup
2379 rd_diag = radiolist_dialog(
2380 values=_radio_values,
2381 title="Scapy v%s" % conf.version,
2382 text=HTML(
2383 (
2384 '<style bg="white" fg="red">Please select a file'
2385 'among the following, to see all layers contained in'
2386 ' it:</style>'
2387 ) + extra_text
2388 ),
2389 cancel_text="Back" if previous else "Cancel"
2390 )
2391 result = call_ptk(rd_diag)
2392 if result is None:
2393 # User pressed "Cancel/Back"
2394 if previous: # Back
2395 current = previous.pop()
2396 continue
2397 else: # Cancel
2398 return
2399 if result.startswith("$"):
2400 previous.append(current)
2401 current = current[result[1:]]
2402 else:
2403 # Enter on layer
2404 if previous: # In subfolder
2405 result = cur_path + "." + result
2406 break
2407 # 4 - (Contrib only): load contrib
2408 if action == "contribs":
2409 from scapy.main import load_contrib
2410 load_contrib(result)
2411 result = "scapy.contrib." + result
2412 else: # NON-GUI MODE
2413 # We handle layer as a short layer name, full layer name
2414 # or the module itself
2415 if isinstance(layer, types.ModuleType):
2416 layer = layer.__name__
2417 if isinstance(layer, str):
2418 if layer.startswith("scapy.layers."):
2419 result = layer
2420 else:
2421 if layer.startswith("scapy.contrib."):
2422 layer = layer.replace("scapy.contrib.", "")
2423 from scapy.main import load_contrib
2424 load_contrib(layer)
2425 result_layer, result_contrib = (("scapy.layers.%s" % layer),
2426 ("scapy.contrib.%s" % layer))
2427 if result_layer in conf.layers.ldict:
2428 result = result_layer
2429 elif result_contrib in conf.layers.ldict:
2430 result = result_contrib
2431 else:
2432 raise Scapy_Exception("Unknown scapy module '%s'" % layer)
2433 else:
2434 warning("Wrong usage ! Check out help(explore)")
2435 return
2437 # COMMON PART
2438 # Get the list of all Packets contained in that module
2439 try:
2440 all_layers = conf.layers.ldict[result]
2441 except KeyError:
2442 raise Scapy_Exception("Unknown scapy module '%s'" % layer)
2443 # Print
2444 print(conf.color_theme.layer_name("Packets contained in %s:" % result))
2445 rtlst = [] # type: List[Tuple[Union[str, List[str]], ...]]
2446 rtlst = [(lay.__name__ or "", cast(str, lay._name) or "") for lay in all_layers]
2447 print(pretty_list(rtlst, [("Class", "Name")], borders=True))
2450def _pkt_ls(obj, # type: Union[Packet, Type[Packet]]
2451 verbose=False, # type: bool
2452 ):
2453 # type: (...) -> List[Tuple[str, Type[AnyField], str, str, List[str]]] # noqa: E501
2454 """Internal function used to resolve `fields_desc` to display it.
2456 :param obj: a packet object or class
2457 :returns: a list containing tuples [(name, clsname, clsname_extras,
2458 default, long_attrs)]
2459 """
2460 is_pkt = isinstance(obj, Packet)
2461 if not issubtype(obj, Packet) and not is_pkt:
2462 raise ValueError
2463 fields = []
2464 for f in obj.fields_desc:
2465 cur_fld = f
2466 attrs = [] # type: List[str]
2467 long_attrs = [] # type: List[str]
2468 while isinstance(cur_fld, (Emph, ConditionalField)):
2469 if isinstance(cur_fld, ConditionalField):
2470 attrs.append(cur_fld.__class__.__name__[:4])
2471 cur_fld = cur_fld.fld
2472 name = cur_fld.name
2473 default = cur_fld.default
2474 if verbose and isinstance(cur_fld, EnumField) \
2475 and hasattr(cur_fld, "i2s") and cur_fld.i2s:
2476 if len(cur_fld.i2s or []) < 50:
2477 long_attrs.extend(
2478 "%s: %d" % (strval, numval)
2479 for numval, strval in
2480 sorted(cur_fld.i2s.items())
2481 )
2482 elif isinstance(cur_fld, MultiEnumField):
2483 if isinstance(obj, Packet):
2484 obj_pkt = obj
2485 else:
2486 obj_pkt = obj()
2487 fld_depend = cur_fld.depends_on(obj_pkt)
2488 attrs.append("Depends on %s" % fld_depend)
2489 if verbose:
2490 cur_i2s = cur_fld.i2s_multi.get(
2491 cur_fld.depends_on(obj_pkt), {}
2492 )
2493 if len(cur_i2s) < 50:
2494 long_attrs.extend(
2495 "%s: %d" % (strval, numval)
2496 for numval, strval in
2497 sorted(cur_i2s.items())
2498 )
2499 elif verbose and isinstance(cur_fld, FlagsField):
2500 names = cur_fld.names
2501 long_attrs.append(", ".join(names))
2502 elif isinstance(cur_fld, MultipleTypeField):
2503 default = cur_fld.dflt.default
2504 attrs.append(", ".join(
2505 x[0].__class__.__name__ for x in
2506 itertools.chain(cur_fld.flds, [(cur_fld.dflt,)])
2507 ))
2509 cls = cur_fld.__class__
2510 class_name_extras = "(%s)" % (
2511 ", ".join(attrs)
2512 ) if attrs else ""
2513 if isinstance(cur_fld, BitField):
2514 class_name_extras += " (%d bit%s)" % (
2515 cur_fld.size,
2516 "s" if cur_fld.size > 1 else ""
2517 )
2518 fields.append(
2519 (name,
2520 cls,
2521 class_name_extras,
2522 repr(default),
2523 long_attrs)
2524 )
2525 return fields
2528@conf.commands.register
2529def ls(obj=None, # type: Optional[Union[str, Packet, Type[Packet]]]
2530 case_sensitive=False, # type: bool
2531 verbose=False # type: bool
2532 ):
2533 # type: (...) -> None
2534 """List available layers, or infos on a given layer class or name.
2536 :param obj: Packet / packet name to use
2537 :param case_sensitive: if obj is a string, is it case sensitive?
2538 :param verbose:
2539 """
2540 if obj is None or isinstance(obj, str):
2541 tip = False
2542 if obj is None:
2543 tip = True
2544 all_layers = sorted(conf.layers, key=lambda x: x.__name__)
2545 else:
2546 pattern = re.compile(
2547 obj,
2548 0 if case_sensitive else re.I
2549 )
2550 # We first order by accuracy, then length
2551 if case_sensitive:
2552 sorter = lambda x: (x.__name__.index(obj), len(x.__name__))
2553 else:
2554 obj = obj.lower()
2555 sorter = lambda x: (x.__name__.lower().index(obj),
2556 len(x.__name__))
2557 all_layers = sorted((layer for layer in conf.layers
2558 if (isinstance(layer.__name__, str) and
2559 pattern.search(layer.__name__)) or
2560 (isinstance(layer.name, str) and
2561 pattern.search(layer.name))),
2562 key=sorter)
2563 for layer in all_layers:
2564 print("%-10s : %s" % (layer.__name__, layer._name))
2565 if tip and conf.interactive:
2566 print("\nTIP: You may use explore() to navigate through all "
2567 "layers using a clear GUI")
2568 else:
2569 try:
2570 fields = _pkt_ls(
2571 obj,
2572 verbose=verbose
2573 )
2574 is_pkt = isinstance(obj, Packet)
2575 # Print
2576 for fname, cls, clsne, dflt, long_attrs in fields:
2577 clsinfo = cls.__name__ + " " + clsne
2578 print("%-10s : %-35s =" % (fname, clsinfo), end=' ')
2579 if is_pkt:
2580 print("%-15r" % (getattr(obj, fname),), end=' ')
2581 print("(%r)" % (dflt,))
2582 for attr in long_attrs:
2583 print("%-15s%s" % ("", attr))
2584 # Restart for payload if any
2585 if is_pkt:
2586 obj = cast(Packet, obj)
2587 if isinstance(obj.payload, NoPayload):
2588 return
2589 print("--")
2590 ls(obj.payload)
2591 except ValueError:
2592 print("Not a packet class or name. Type 'ls()' to list packet classes.") # noqa: E501
2595@conf.commands.register
2596def rfc(cls, ret=False, legend=True):
2597 # type: (Type[Packet], bool, bool) -> Optional[str]
2598 """
2599 Generate an RFC-like representation of a packet def.
2601 :param cls: the Packet class
2602 :param ret: return the result instead of printing (def. False)
2603 :param legend: show text under the diagram (default True)
2605 Ex::
2607 >>> rfc(Ether)
2609 """
2610 if not issubclass(cls, Packet):
2611 raise TypeError("Packet class expected")
2612 cur_len = 0
2613 cur_line = []
2614 lines = []
2615 # Get the size (width) that a field will take
2616 # when formatted, from its length in bits
2617 clsize = lambda x: 2 * x - 1 # type: Callable[[int], int]
2618 ident = 0 # Fields UUID
2620 # Generate packet groups
2621 def _iterfields() -> Iterator[Tuple[str, int]]:
2622 for f in cls.fields_desc:
2623 # Fancy field name
2624 fname = f.name.upper().replace("_", " ")
2625 fsize = int(f.sz * 8)
2626 yield fname, fsize
2627 # Add padding optionally
2628 if isinstance(f, PadField):
2629 if isinstance(f._align, tuple):
2630 pad = - cur_len % (f._align[0] * 8)
2631 else:
2632 pad = - cur_len % (f._align * 8)
2633 if pad:
2634 yield "padding", pad
2635 for fname, flen in _iterfields():
2636 cur_len += flen
2637 ident += 1
2638 # The field might exceed the current line or
2639 # take more than one line. Copy it as required
2640 while True:
2641 over = max(0, cur_len - 32) # Exceed
2642 len1 = clsize(flen - over) # What fits
2643 cur_line.append((fname[:len1], len1, ident))
2644 if cur_len >= 32:
2645 # Current line is full. start a new line
2646 lines.append(cur_line)
2647 cur_len = flen = over
2648 fname = "" # do not repeat the field
2649 cur_line = []
2650 if not over:
2651 # there is no data left
2652 break
2653 else:
2654 # End of the field
2655 break
2656 # Add the last line if un-finished
2657 if cur_line:
2658 lines.append(cur_line)
2659 # Calculate separations between lines
2660 seps = []
2661 seps.append("+-" * 32 + "+\n")
2662 for i in range(len(lines) - 1):
2663 # Start with a full line
2664 sep = "+-" * 32 + "+\n"
2665 # Get the line above and below the current
2666 # separation
2667 above, below = lines[i], lines[i + 1]
2668 # The last field of above is shared with below
2669 if above[-1][2] == below[0][2]:
2670 # where the field in "above" starts
2671 pos_above = sum(x[1] for x in above[:-1]) + len(above[:-1]) - 1
2672 # where the field in "below" ends
2673 pos_below = below[0][1]
2674 if pos_above < pos_below:
2675 # they are overlapping.
2676 # Now crop the space between those pos
2677 # and fill it with " "
2678 pos_above = pos_above + pos_above % 2
2679 sep = (
2680 sep[:1 + pos_above] +
2681 " " * (pos_below - pos_above) +
2682 sep[1 + pos_below:]
2683 )
2684 # line is complete
2685 seps.append(sep)
2686 # Graph
2687 result = ""
2688 # Bytes markers
2689 result += " " + (" " * 19).join(
2690 str(x) for x in range(4)
2691 ) + "\n"
2692 # Bits markers
2693 result += " " + " ".join(
2694 str(x % 10) for x in range(32)
2695 ) + "\n"
2696 # Add fields and their separations
2697 for line, sep in zip(lines, seps):
2698 result += sep
2699 for elt, flen, _ in line:
2700 result += "|" + elt.center(flen, " ")
2701 result += "|\n"
2702 result += "+-" * (cur_len or 32) + "+\n"
2703 # Annotate with the figure name
2704 if legend:
2705 result += "\n" + ("Fig. " + cls.__name__).center(66, " ")
2706 # return if asked for, else print
2707 if ret:
2708 return result
2709 print(result)
2710 return None
2713#############
2714# Fuzzing #
2715#############
2717_P = TypeVar('_P', bound=Packet)
2720@conf.commands.register
2721def fuzz(p, # type: _P
2722 _inplace=0, # type: int
2723 ):
2724 # type: (...) -> _P
2725 """
2726 Transform a layer into a fuzzy layer by replacing some default values
2727 by random objects.
2729 :param p: the Packet instance to fuzz
2730 :return: the fuzzed packet.
2731 """
2732 if not _inplace:
2733 p = p.copy()
2734 q = cast(Packet, p)
2735 while not isinstance(q, NoPayload):
2736 new_default_fields = {}
2737 multiple_type_fields = [] # type: List[str]
2738 for f in q.fields_desc:
2739 if isinstance(f, PacketListField):
2740 for r in getattr(q, f.name):
2741 fuzz(r, _inplace=1)
2742 elif isinstance(f, MultipleTypeField):
2743 # the type of the field will depend on others
2744 multiple_type_fields.append(f.name)
2745 elif f.default is not None:
2746 if not isinstance(f, ConditionalField) or f._evalcond(q):
2747 rnd = f.randval()
2748 if rnd is not None:
2749 new_default_fields[f.name] = rnd
2750 # Process packets with MultipleTypeFields
2751 if multiple_type_fields:
2752 # freeze the other random values
2753 new_default_fields = {
2754 key: (val._fix() if isinstance(val, VolatileValue) else val)
2755 for key, val in new_default_fields.items()
2756 }
2757 q.default_fields.update(new_default_fields)
2758 new_default_fields.clear()
2759 # add the random values of the MultipleTypeFields
2760 for name in multiple_type_fields:
2761 fld = cast(MultipleTypeField, q.get_field(name))
2762 rnd = fld._find_fld_pkt(q).randval()
2763 if rnd is not None:
2764 new_default_fields[name] = rnd
2765 q.default_fields.update(new_default_fields)
2766 q = q.payload
2767 return p