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