Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/fields.py: 70%
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>
5# Copyright (C) Michael Farrell <micolous+git@gmail.com>
8"""
9Fields: basic data structures that make up parts of packets.
10"""
12import calendar
13import collections
14import copy
15import datetime
16import inspect
17import math
18import socket
19import struct
20import time
21import warnings
23from types import MethodType
24from uuid import UUID
25from enum import Enum
27from scapy.config import conf
28from scapy.dadict import DADict
29from scapy.volatile import RandBin, RandByte, RandEnumKeys, RandInt, \
30 RandIP, RandIP6, RandLong, RandMAC, RandNum, RandShort, RandSInt, \
31 RandSByte, RandTermString, RandUUID, VolatileValue, RandSShort, \
32 RandSLong, RandFloat
33from scapy.data import EPOCH
34from scapy.error import log_runtime, Scapy_Exception
35from scapy.compat import bytes_hex, plain_str, raw, bytes_encode
36from scapy.pton_ntop import inet_ntop, inet_pton
37from scapy.utils import inet_aton, inet_ntoa, lhex, mac2str, str2mac, EDecimal
38from scapy.utils6 import in6_6to4ExtractAddr, in6_isaddr6to4, \
39 in6_isaddrTeredo, in6_ptop, Net6, teredoAddrExtractInfo
40from scapy.base_classes import (
41 _ScopedIP,
42 BasePacket,
43 Field_metaclass,
44 Net,
45 ScopedIP,
46)
48# Typing imports
49from typing import (
50 Any,
51 AnyStr,
52 Callable,
53 Dict,
54 List,
55 Generic,
56 Optional,
57 Set,
58 Tuple,
59 Type,
60 TypeVar,
61 Union,
62 # func
63 cast,
64 TYPE_CHECKING,
65)
67if TYPE_CHECKING:
68 # Do not import on runtime ! (import loop)
69 from scapy.packet import Packet
72class RawVal:
73 r"""
74 A raw value that will not be processed by the field and inserted
75 as-is in the packet string.
77 Example::
79 >>> a = IP(len=RawVal("####"))
80 >>> bytes(a)
81 b'F\x00####\x00\x01\x00\x005\xb5\x00\x00\x7f\x00\x00\x01\x7f\x00\x00\x01\x00\x00'
83 """
85 def __init__(self, val=b""):
86 # type: (bytes) -> None
87 self.val = bytes_encode(val)
89 def __str__(self):
90 # type: () -> str
91 return str(self.val)
93 def __bytes__(self):
94 # type: () -> bytes
95 return self.val
97 def __len__(self):
98 # type: () -> int
99 return len(self.val)
101 def __repr__(self):
102 # type: () -> str
103 return "<RawVal [%r]>" % self.val
106class ObservableDict(Dict[int, str]):
107 """
108 Helper class to specify a protocol extendable for runtime modifications
109 """
111 def __init__(self, *args, **kw):
112 # type: (*Dict[int, str], **Any) -> None
113 self.observers = [] # type: List[_EnumField[Any]]
114 super(ObservableDict, self).__init__(*args, **kw)
116 def observe(self, observer):
117 # type: (_EnumField[Any]) -> None
118 self.observers.append(observer)
120 def __setitem__(self, key, value):
121 # type: (int, str) -> None
122 for o in self.observers:
123 o.notify_set(self, key, value)
124 super(ObservableDict, self).__setitem__(key, value)
126 def __delitem__(self, key):
127 # type: (int) -> None
128 for o in self.observers:
129 o.notify_del(self, key)
130 super(ObservableDict, self).__delitem__(key)
132 def update(self, anotherDict): # type: ignore
133 for k in anotherDict:
134 self[k] = anotherDict[k]
137############
138# Fields #
139############
141I = TypeVar('I') # Internal storage # noqa: E741
142M = TypeVar('M') # Machine storage
145class Field(Generic[I, M], metaclass=Field_metaclass):
146 """
147 For more information on how this works, please refer to the
148 'Adding new protocols' chapter in the online documentation:
150 https://scapy.readthedocs.io/en/stable/build_dissect.html
151 """
152 __slots__ = [
153 "name",
154 "fmt",
155 "default",
156 "sz",
157 "owners",
158 "struct"
159 ]
160 islist = 0
161 ismutable = False
162 holds_packets = 0
163 isconditional = False
164 ismayend = False
166 def __init__(self, name, default, fmt="H"):
167 # type: (str, Any, str) -> None
168 if not isinstance(name, str):
169 raise ValueError("name should be a string")
170 self.name = name
171 if fmt[0] in "@=<>!":
172 self.fmt = fmt
173 else:
174 self.fmt = "!" + fmt
175 self.struct = struct.Struct(self.fmt)
176 self.default = self.any2i(None, default)
177 self.sz = struct.calcsize(self.fmt) # type: int
178 self.owners = [] # type: List[Type[Packet]]
180 def register_owner(self, cls):
181 # type: (Type[Packet]) -> None
182 self.owners.append(cls)
184 def i2len(self,
185 pkt, # type: Packet
186 x, # type: Any
187 ):
188 # type: (...) -> int
189 """Convert internal value to a length usable by a FieldLenField"""
190 if isinstance(x, RawVal):
191 return len(x)
192 return self.sz
194 def i2count(self, pkt, x):
195 # type: (Optional[Packet], I) -> int
196 """Convert internal value to a number of elements usable by a FieldLenField.
197 Always 1 except for list fields"""
198 return 1
200 def h2i(self, pkt, x):
201 # type: (Optional[Packet], Any) -> I
202 """Convert human value to internal value"""
203 return x # type: ignore
205 def i2h(self, pkt, x):
206 # type: (Optional[Packet], I) -> Any
207 """Convert internal value to human value"""
208 return x
210 def m2i(self, pkt, x):
211 # type: (Optional[Packet], M) -> I
212 """Convert machine value to internal value"""
213 return x # type: ignore
215 def i2m(self, pkt, x):
216 # type: (Optional[Packet], Optional[I]) -> M
217 """Convert internal value to machine value"""
218 if x is None:
219 return 0 # type: ignore
220 elif isinstance(x, str):
221 return bytes_encode(x) # type: ignore
222 return x # type: ignore
224 def any2i(self, pkt, x):
225 # type: (Optional[Packet], Any) -> Optional[I]
226 """Try to understand the most input values possible and make an internal value from them""" # noqa: E501
227 return self.h2i(pkt, x)
229 def i2repr(self, pkt, x):
230 # type: (Optional[Packet], I) -> str
231 """Convert internal value to a nice representation"""
232 return repr(self.i2h(pkt, x))
234 def addfield(self, pkt, s, val):
235 # type: (Packet, bytes, Optional[I]) -> bytes
236 """Add an internal value to a string
238 Copy the network representation of field `val` (belonging to layer
239 `pkt`) to the raw string packet `s`, and return the new string packet.
240 """
241 try:
242 return s + self.struct.pack(self.i2m(pkt, val))
243 except struct.error as ex:
244 raise ValueError(
245 "Incorrect type of value for field %s:\n" % self.name +
246 "struct.error('%s')\n" % ex +
247 "To inject bytes into the field regardless of the type, " +
248 "use RawVal. See help(RawVal)"
249 )
251 def getfield(self, pkt, s):
252 # type: (Packet, bytes) -> Tuple[bytes, I]
253 """Extract an internal value from a string
255 Extract from the raw packet `s` the field value belonging to layer
256 `pkt`.
258 Returns a two-element list,
259 first the raw packet string after having removed the extracted field,
260 second the extracted field itself in internal representation.
261 """
262 # Use unpack_from for plain bytes (avoids temporary slice allocation).
263 # Fall back to unpack+slice for subclasses that override __getitem__.
264 if type(s) is bytes:
265 return s[self.sz:], self.m2i(pkt, self.struct.unpack_from(s)[0])
266 return s[self.sz:], self.m2i(pkt, self.struct.unpack(s[:self.sz])[0])
268 def do_copy(self, x):
269 # type: (I) -> I
270 if isinstance(x, list):
271 x = x[:] # type: ignore
272 for i in range(len(x)):
273 if isinstance(x[i], BasePacket):
274 x[i] = x[i].copy()
275 return x # type: ignore
276 if hasattr(x, "copy"):
277 return x.copy() # type: ignore
278 return x
280 def __repr__(self):
281 # type: () -> str
282 return "<%s (%s).%s>" % (
283 self.__class__.__name__,
284 ",".join(x.__name__ for x in self.owners),
285 self.name
286 )
288 def copy(self):
289 # type: () -> Field[I, M]
290 return copy.copy(self)
292 def randval(self):
293 # type: () -> VolatileValue[Any]
294 """Return a volatile object whose value is both random and suitable for this field""" # noqa: E501
295 fmtt = self.fmt[-1]
296 if fmtt in "BbHhIiQq":
297 return {"B": RandByte, "b": RandSByte,
298 "H": RandShort, "h": RandSShort,
299 "I": RandInt, "i": RandSInt,
300 "Q": RandLong, "q": RandSLong}[fmtt]()
301 elif fmtt == "s":
302 if self.fmt[0] in "0123456789":
303 value = int(self.fmt[:-1])
304 else:
305 value = int(self.fmt[1:-1])
306 return RandBin(value)
307 else:
308 raise ValueError(
309 "no random class for [%s] (fmt=%s)." % (
310 self.name, self.fmt
311 )
312 )
315class _FieldContainer(object):
316 """
317 A field that acts as a container for another field
318 """
319 __slots__ = ["fld"]
320 isconditional = False
321 ismayend = False
323 def __getattr__(self, attr):
324 # type: (str) -> Any
325 return getattr(self.fld, attr)
328AnyField = Union[Field[Any, Any], _FieldContainer]
331class Emph(_FieldContainer):
332 """Empathize sub-layer for display"""
333 __slots__ = ["fld"]
335 def __init__(self, fld):
336 # type: (Any) -> None
337 self.fld = fld
339 def __eq__(self, other):
340 # type: (Any) -> bool
341 return bool(self.fld == other)
343 def __hash__(self):
344 # type: () -> int
345 return hash(self.fld)
348class MayEnd(_FieldContainer):
349 """
350 Allow packet dissection to end after the dissection of this field
351 if no bytes are left.
353 A good example would be a length field that can be 0 or a set value,
354 and where it would be too annoying to use multiple ConditionalFields
356 Important note: any field below this one MUST default
357 to an empty value, else the behavior will be unexpected.
358 """
359 __slots__ = ["fld"]
360 ismayend = True
362 def __init__(self, fld):
363 # type: (Any) -> None
364 self.fld = fld
366 def __eq__(self, other):
367 # type: (Any) -> bool
368 return bool(self.fld == other)
370 def __hash__(self):
371 # type: () -> int
372 return hash(self.fld)
375class ActionField(_FieldContainer):
376 __slots__ = ["fld", "_action_method", "_privdata"]
378 def __init__(self, fld, action_method, **kargs):
379 # type: (Field[Any, Any], str, **Any) -> None
380 self.fld = fld
381 self._action_method = action_method
382 self._privdata = kargs
384 def any2i(self, pkt, val):
385 # type: (Optional[Packet], int) -> Any
386 getattr(pkt, self._action_method)(val, self.fld, **self._privdata)
387 return getattr(self.fld, "any2i")(pkt, val)
390class ConditionalField(_FieldContainer):
391 __slots__ = ["fld", "cond"]
392 isconditional = True
394 def __init__(self,
395 fld, # type: AnyField
396 cond # type: Callable[[Packet], bool]
397 ):
398 # type: (...) -> None
399 self.fld = fld
400 self.cond = cond
402 def _evalcond(self, pkt):
403 # type: (Packet) -> bool
404 return bool(self.cond(pkt))
406 def any2i(self, pkt, x):
407 # type: (Optional[Packet], Any) -> Any
408 # BACKWARD COMPATIBILITY
409 # Note: we shouldn't need this function. (it's not correct)
410 # However, having i2h implemented (#2364), it changes the default
411 # behavior and broke all packets that wrongly use two ConditionalField
412 # with the same name. Those packets are the problem: they are wrongly
413 # built (they should either be reusing the same conditional field, or
414 # using a MultipleTypeField).
415 # But I don't want to dive into fixing all of them just yet,
416 # so for now, let's keep this this way, even though it's not correct.
417 if type(self.fld) is Field:
418 return x
419 return self.fld.any2i(pkt, x)
421 def i2h(self, pkt, val):
422 # type: (Optional[Packet], Any) -> Any
423 if pkt and not self._evalcond(pkt):
424 return None
425 return self.fld.i2h(pkt, val)
427 def getfield(self, pkt, s):
428 # type: (Packet, bytes) -> Tuple[bytes, Any]
429 if self._evalcond(pkt):
430 return self.fld.getfield(pkt, s)
431 else:
432 return s, None
434 def addfield(self, pkt, s, val):
435 # type: (Packet, bytes, Any) -> bytes
436 if self._evalcond(pkt):
437 return self.fld.addfield(pkt, s, val)
438 else:
439 return s
441 def __getattr__(self, attr):
442 # type: (str) -> Any
443 return getattr(self.fld, attr)
446class MultipleTypeField(_FieldContainer):
447 """
448 MultipleTypeField are used for fields that can be implemented by
449 various Field subclasses, depending on conditions on the packet.
451 It is initialized with `flds` and `dflt`.
453 :param dflt: is the default field type, to be used when none of the
454 conditions matched the current packet.
455 :param flds: is a list of tuples (`fld`, `cond`) or (`fld`, `cond`, `hint`)
456 where `fld` if a field type, and `cond` a "condition" to
457 determine if `fld` is the field type that should be used.
459 ``cond`` is either:
461 - a callable `cond_pkt` that accepts one argument (the packet) and
462 returns True if `fld` should be used, False otherwise.
463 - a tuple (`cond_pkt`, `cond_pkt_val`), where `cond_pkt` is the same
464 as in the previous case and `cond_pkt_val` is a callable that
465 accepts two arguments (the packet, and the value to be set) and
466 returns True if `fld` should be used, False otherwise.
468 See scapy.layers.l2.ARP (type "help(ARP)" in Scapy) for an example of
469 use.
470 """
472 __slots__ = ["flds", "dflt", "hints", "name", "default"]
474 def __init__(
475 self,
476 flds: List[Union[
477 Tuple[Field[Any, Any], Any, str],
478 Tuple[Field[Any, Any], Any]
479 ]],
480 dflt: Field[Any, Any]
481 ) -> None:
482 self.hints = {
483 x[0]: x[2]
484 for x in flds
485 if len(x) == 3
486 }
487 self.flds = [
488 (x[0], x[1]) for x in flds
489 ]
490 self.dflt = dflt
491 self.default = None # So that we can detect changes in defaults
492 self.name = self.dflt.name
493 if any(x[0].name != self.name for x in self.flds):
494 warnings.warn(
495 ("All fields should have the same name in a "
496 "MultipleTypeField (%s). Use hints.") % self.name,
497 SyntaxWarning
498 )
500 def _iterate_fields_cond(self, pkt, val, use_val):
501 # type: (Optional[Packet], Any, bool) -> Field[Any, Any]
502 """Internal function used by _find_fld_pkt & _find_fld_pkt_val"""
503 # Iterate through the fields
504 for fld, cond in self.flds:
505 if isinstance(cond, tuple):
506 if use_val:
507 if val is None:
508 val = self.dflt.default
509 if cond[1](pkt, val):
510 return fld
511 continue
512 else:
513 cond = cond[0]
514 if cond(pkt):
515 return fld
516 return self.dflt
518 def _find_fld_pkt(self, pkt):
519 # type: (Optional[Packet]) -> Field[Any, Any]
520 """Given a Packet instance `pkt`, returns the Field subclass to be
521used. If you know the value to be set (e.g., in .addfield()), use
522._find_fld_pkt_val() instead.
524 """
525 return self._iterate_fields_cond(pkt, None, False)
527 def _find_fld_pkt_val(self,
528 pkt, # type: Optional[Packet]
529 val, # type: Any
530 ):
531 # type: (...) -> Tuple[Field[Any, Any], Any]
532 """Given a Packet instance `pkt` and the value `val` to be set,
533returns the Field subclass to be used, and the updated `val` if necessary.
535 """
536 fld = self._iterate_fields_cond(pkt, val, True)
537 if val is None:
538 val = fld.default
539 return fld, val
541 def _find_fld(self):
542 # type: () -> Field[Any, Any]
543 """Returns the Field subclass to be used, depending on the Packet
544instance, or the default subclass.
546DEV: since the Packet instance is not provided, we have to use a hack
547to guess it. It should only be used if you cannot provide the current
548Packet instance (for example, because of the current Scapy API).
550If you have the current Packet instance, use ._find_fld_pkt_val() (if
551the value to set is also known) of ._find_fld_pkt() instead.
553 """
554 # Hack to preserve current Scapy API
555 # See https://stackoverflow.com/a/7272464/3223422
556 frame = inspect.currentframe().f_back.f_back # type: ignore
557 while frame is not None:
558 try:
559 pkt = frame.f_locals['self']
560 except KeyError:
561 pass
562 else:
563 if isinstance(pkt, tuple(self.dflt.owners)):
564 if not pkt.default_fields:
565 # Packet not initialized
566 return self.dflt
567 return self._find_fld_pkt(pkt)
568 frame = frame.f_back
569 return self.dflt
571 def getfield(self,
572 pkt, # type: Packet
573 s, # type: bytes
574 ):
575 # type: (...) -> Tuple[bytes, Any]
576 return self._find_fld_pkt(pkt).getfield(pkt, s)
578 def addfield(self, pkt, s, val):
579 # type: (Packet, bytes, Any) -> bytes
580 fld, val = self._find_fld_pkt_val(pkt, val)
581 return fld.addfield(pkt, s, val)
583 def any2i(self, pkt, val):
584 # type: (Optional[Packet], Any) -> Any
585 fld, val = self._find_fld_pkt_val(pkt, val)
586 return fld.any2i(pkt, val)
588 def h2i(self, pkt, val):
589 # type: (Optional[Packet], Any) -> Any
590 fld, val = self._find_fld_pkt_val(pkt, val)
591 return fld.h2i(pkt, val)
593 def i2h(self,
594 pkt, # type: Packet
595 val, # type: Any
596 ):
597 # type: (...) -> Any
598 fld, val = self._find_fld_pkt_val(pkt, val)
599 return fld.i2h(pkt, val)
601 def i2m(self, pkt, val):
602 # type: (Optional[Packet], Optional[Any]) -> Any
603 fld, val = self._find_fld_pkt_val(pkt, val)
604 return fld.i2m(pkt, val)
606 def i2len(self, pkt, val):
607 # type: (Packet, Any) -> int
608 fld, val = self._find_fld_pkt_val(pkt, val)
609 return fld.i2len(pkt, val)
611 def i2repr(self, pkt, val):
612 # type: (Optional[Packet], Any) -> str
613 fld, val = self._find_fld_pkt_val(pkt, val)
614 hint = ""
615 if fld in self.hints:
616 hint = " (%s)" % self.hints[fld]
617 return fld.i2repr(pkt, val) + hint
619 def register_owner(self, cls):
620 # type: (Type[Packet]) -> None
621 for fld, _ in self.flds:
622 fld.owners.append(cls)
623 self.dflt.owners.append(cls)
625 def get_fields_list(self):
626 # type: () -> List[Any]
627 return [self]
629 @property
630 def fld(self):
631 # type: () -> Field[Any, Any]
632 return self._find_fld()
635class PadField(_FieldContainer):
636 """Add bytes after the proxified field so that it ends at the specified
637 alignment from its beginning"""
638 __slots__ = ["fld", "_align", "_padwith"]
640 def __init__(self, fld, align, padwith=None):
641 # type: (AnyField, int, Optional[bytes]) -> None
642 self.fld = fld
643 self._align = align
644 self._padwith = padwith or b"\x00"
646 def padlen(self, flen, pkt):
647 # type: (int, Packet) -> int
648 return -flen % self._align
650 def getfield(self,
651 pkt, # type: Packet
652 s, # type: bytes
653 ):
654 # type: (...) -> Tuple[bytes, Any]
655 remain, val = self.fld.getfield(pkt, s)
656 padlen = self.padlen(len(s) - len(remain), pkt)
657 return remain[padlen:], val
659 def addfield(self,
660 pkt, # type: Packet
661 s, # type: bytes
662 val, # type: Any
663 ):
664 # type: (...) -> bytes
665 sval = self.fld.addfield(pkt, b"", val)
666 return s + sval + (
667 self.padlen(len(sval), pkt) * self._padwith
668 )
671class ReversePadField(PadField):
672 """Add bytes BEFORE the proxified field so that it starts at the specified
673 alignment from its beginning"""
675 def original_length(self, pkt):
676 # type: (Packet) -> int
677 return len(pkt.original)
679 def getfield(self,
680 pkt, # type: Packet
681 s, # type: bytes
682 ):
683 # type: (...) -> Tuple[bytes, Any]
684 # We need to get the length that has already been dissected
685 padlen = self.padlen(self.original_length(pkt) - len(s), pkt)
686 return self.fld.getfield(pkt, s[padlen:])
688 def addfield(self,
689 pkt, # type: Packet
690 s, # type: bytes
691 val, # type: Any
692 ):
693 # type: (...) -> bytes
694 sval = self.fld.addfield(pkt, b"", val)
695 return s + struct.pack("%is" % (
696 self.padlen(len(s), pkt)
697 ), self._padwith) + sval
700class TrailerBytes(bytes):
701 """
702 Reverses slice operations to take from the back of the packet,
703 not the front
704 """
706 def __getitem__(self, item): # type: ignore
707 # type: (Union[int, slice]) -> Union[int, bytes]
708 if isinstance(item, int):
709 if item < 0:
710 item = 1 + item
711 else:
712 item = len(self) - 1 - item
713 elif isinstance(item, slice):
714 start, stop, step = item.start, item.stop, item.step
715 new_start = -stop if stop else None
716 new_stop = -start if start else None
717 item = slice(new_start, new_stop, step)
718 return super(self.__class__, self).__getitem__(item)
721class TrailerField(_FieldContainer):
722 """Special Field that gets its value from the end of the *packet*
723 (Note: not layer, but packet).
725 Mostly used for FCS
726 """
727 __slots__ = ["fld"]
729 def __init__(self, fld):
730 # type: (Field[Any, Any]) -> None
731 self.fld = fld
733 # Note: this is ugly. Very ugly.
734 # Do not copy this crap elsewhere, so that if one day we get
735 # brave enough to refactor it, it'll be easier.
737 def getfield(self, pkt, s):
738 # type: (Packet, bytes) -> Tuple[bytes, int]
739 previous_post_dissect = pkt.post_dissect
741 def _post_dissect(self, s):
742 # type: (Packet, bytes) -> bytes
743 # Reset packet to allow post_build
744 self.raw_packet_cache = None
745 self.post_dissect = previous_post_dissect # type: ignore
746 return previous_post_dissect(s)
747 pkt.post_dissect = MethodType(_post_dissect, pkt) # type: ignore
748 s = TrailerBytes(s)
749 s, val = self.fld.getfield(pkt, s)
750 return bytes(s), val
752 def addfield(self, pkt, s, val):
753 # type: (Packet, bytes, Optional[int]) -> bytes
754 previous_post_build = pkt.post_build
755 value = self.fld.addfield(pkt, b"", val)
757 def _post_build(self, p, pay):
758 # type: (Packet, bytes, bytes) -> bytes
759 pay += value
760 self.post_build = previous_post_build # type: ignore
761 return previous_post_build(p, pay)
762 pkt.post_build = MethodType(_post_build, pkt) # type: ignore
763 return s
766class FCSField(TrailerField):
767 """
768 A FCS field that gets appended at the end of the *packet* (not layer).
769 """
771 def __init__(self, *args, **kwargs):
772 # type: (*Any, **Any) -> None
773 super(FCSField, self).__init__(Field(*args, **kwargs))
775 def i2repr(self, pkt, x):
776 # type: (Optional[Packet], int) -> str
777 return lhex(self.i2h(pkt, x))
780class DestField(Field[str, bytes]):
781 __slots__ = ["defaultdst"]
782 # Each subclass must have its own bindings attribute
783 bindings = {} # type: Dict[Type[Packet], Tuple[str, Any]]
785 def __init__(self, name, default):
786 # type: (str, str) -> None
787 self.defaultdst = default
789 def dst_from_pkt(self, pkt):
790 # type: (Packet) -> str
791 for addr, condition in self.bindings.get(pkt.payload.__class__, []):
792 try:
793 if all(pkt.payload.getfieldval(field) == value
794 for field, value in condition.items()):
795 if callable(addr):
796 return addr(pkt) # type: ignore
797 else:
798 return addr # type: ignore
799 except AttributeError:
800 pass
801 return self.defaultdst
803 @classmethod
804 def bind_addr(cls, layer, addr, **condition):
805 # type: (Type[Packet], str, **Any) -> None
806 cls.bindings.setdefault(layer, []).append( # type: ignore
807 (addr, condition)
808 )
811class MACField(Field[Optional[str], bytes]):
812 def __init__(self, name, default):
813 # type: (str, Optional[Any]) -> None
814 Field.__init__(self, name, default, "6s")
816 def i2m(self, pkt, x):
817 # type: (Optional[Packet], Optional[str]) -> bytes
818 if not x:
819 return b"\0\0\0\0\0\0"
820 try:
821 y = mac2str(x)
822 except (struct.error, OverflowError, ValueError):
823 y = bytes_encode(x)
824 return y
826 def m2i(self, pkt, x):
827 # type: (Optional[Packet], bytes) -> str
828 return str2mac(x)
830 def any2i(self, pkt, x):
831 # type: (Optional[Packet], Any) -> str
832 if isinstance(x, bytes) and len(x) == 6:
833 return self.m2i(pkt, x)
834 return cast(str, x)
836 def i2repr(self, pkt, x):
837 # type: (Optional[Packet], Optional[str]) -> str
838 x = self.i2h(pkt, x)
839 if x is None:
840 return repr(x)
841 if self in conf.resolve:
842 x = conf.manufdb._resolve_MAC(x)
843 return x
845 def randval(self):
846 # type: () -> RandMAC
847 return RandMAC()
850class LEMACField(MACField):
851 def i2m(self, pkt, x):
852 # type: (Optional[Packet], Optional[str]) -> bytes
853 return MACField.i2m(self, pkt, x)[::-1]
855 def m2i(self, pkt, x):
856 # type: (Optional[Packet], bytes) -> str
857 return MACField.m2i(self, pkt, x[::-1])
860class IPField(Field[Union[str, Net], bytes]):
861 def __init__(self, name, default):
862 # type: (str, Optional[str]) -> None
863 Field.__init__(self, name, default, "4s")
865 def h2i(self, pkt, x):
866 # type: (Optional[Packet], Union[AnyStr, List[AnyStr]]) -> Any
867 if isinstance(x, bytes):
868 x = plain_str(x) # type: ignore
869 if isinstance(x, _ScopedIP):
870 return x
871 elif isinstance(x, str):
872 x = ScopedIP(x)
873 try:
874 inet_aton(x)
875 except socket.error:
876 return Net(x)
877 elif isinstance(x, tuple):
878 if len(x) != 2:
879 raise ValueError("Invalid IP format")
880 return Net(*x)
881 elif isinstance(x, list):
882 return [self.h2i(pkt, n) for n in x]
883 return x
885 def i2h(self, pkt, x):
886 # type: (Optional[Packet], Optional[Union[str, Net]]) -> str
887 return cast(str, x)
889 def resolve(self, x):
890 # type: (str) -> str
891 if self in conf.resolve:
892 try:
893 ret = socket.gethostbyaddr(x)[0]
894 except Exception:
895 pass
896 else:
897 if ret:
898 return ret
899 return x
901 def i2m(self, pkt, x):
902 # type: (Optional[Packet], Optional[Union[str, Net]]) -> bytes
903 if x is None:
904 return b'\x00\x00\x00\x00'
905 return inet_aton(plain_str(x))
907 def m2i(self, pkt, x):
908 # type: (Optional[Packet], bytes) -> str
909 return inet_ntoa(x)
911 def any2i(self, pkt, x):
912 # type: (Optional[Packet], Any) -> Any
913 return self.h2i(pkt, x)
915 def i2repr(self, pkt, x):
916 # type: (Optional[Packet], Union[str, Net]) -> str
917 if isinstance(x, _ScopedIP) and x.scope:
918 return repr(x)
919 r = self.resolve(self.i2h(pkt, x))
920 return r if isinstance(r, str) else repr(r)
922 def randval(self):
923 # type: () -> RandIP
924 return RandIP()
927class SourceIPField(IPField):
928 def __init__(self, name):
929 # type: (str) -> None
930 IPField.__init__(self, name, None)
932 def __findaddr(self, pkt):
933 # type: (Packet) -> Optional[str]
934 if conf.route is None:
935 # unused import, only to initialize conf.route
936 import scapy.route # noqa: F401
937 return pkt.route()[1] or conf.route.route()[1]
939 def i2m(self, pkt, x):
940 # type: (Optional[Packet], Optional[Union[str, Net]]) -> bytes
941 if x is None and pkt is not None:
942 x = self.__findaddr(pkt)
943 return super(SourceIPField, self).i2m(pkt, x)
945 def i2h(self, pkt, x):
946 # type: (Optional[Packet], Optional[Union[str, Net]]) -> str
947 if x is None and pkt is not None:
948 x = self.__findaddr(pkt)
949 return super(SourceIPField, self).i2h(pkt, x)
952class IP6Field(Field[Optional[Union[str, Net6]], bytes]):
953 def __init__(self, name, default):
954 # type: (str, Optional[str]) -> None
955 Field.__init__(self, name, default, "16s")
957 def h2i(self, pkt, x):
958 # type: (Optional[Packet], Any) -> str
959 if isinstance(x, bytes):
960 x = plain_str(x)
961 if isinstance(x, _ScopedIP):
962 return x
963 elif isinstance(x, str):
964 x = ScopedIP(x)
965 try:
966 x = ScopedIP(in6_ptop(x), scope=x.scope)
967 except socket.error:
968 return Net6(x) # type: ignore
969 elif isinstance(x, tuple):
970 if len(x) != 2:
971 raise ValueError("Invalid IPv6 format")
972 return Net6(*x) # type: ignore
973 elif isinstance(x, list):
974 x = [self.h2i(pkt, n) for n in x]
975 return x # type: ignore
977 def i2h(self, pkt, x):
978 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> str
979 return cast(str, x)
981 def i2m(self, pkt, x):
982 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> bytes
983 if x is None:
984 x = "::"
985 return inet_pton(socket.AF_INET6, plain_str(x))
987 def m2i(self, pkt, x):
988 # type: (Optional[Packet], bytes) -> str
989 return inet_ntop(socket.AF_INET6, x)
991 def any2i(self, pkt, x):
992 # type: (Optional[Packet], Optional[str]) -> str
993 return self.h2i(pkt, x)
995 def i2repr(self, pkt, x):
996 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> str
997 if x is None:
998 return self.i2h(pkt, x)
999 elif not isinstance(x, Net6) and not isinstance(x, list):
1000 if in6_isaddrTeredo(x): # print Teredo info
1001 server, _, maddr, mport = teredoAddrExtractInfo(x)
1002 return "%s [Teredo srv: %s cli: %s:%s]" % (self.i2h(pkt, x), server, maddr, mport) # noqa: E501
1003 elif in6_isaddr6to4(x): # print encapsulated address
1004 vaddr = in6_6to4ExtractAddr(x)
1005 return "%s [6to4 GW: %s]" % (self.i2h(pkt, x), vaddr)
1006 elif isinstance(x, _ScopedIP) and x.scope:
1007 return repr(x)
1008 r = self.i2h(pkt, x) # No specific information to return
1009 return r if isinstance(r, str) else repr(r)
1011 def randval(self):
1012 # type: () -> RandIP6
1013 return RandIP6()
1016class SourceIP6Field(IP6Field):
1017 def __init__(self, name):
1018 # type: (str) -> None
1019 IP6Field.__init__(self, name, None)
1021 def __findaddr(self, pkt):
1022 # type: (Packet) -> Optional[str]
1023 if conf.route6 is None:
1024 # unused import, only to initialize conf.route
1025 import scapy.route6 # noqa: F401
1026 return pkt.route()[1]
1028 def i2m(self, pkt, x):
1029 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> bytes
1030 if x is None and pkt is not None:
1031 x = self.__findaddr(pkt)
1032 return super(SourceIP6Field, self).i2m(pkt, x)
1034 def i2h(self, pkt, x):
1035 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> str
1036 if x is None and pkt is not None:
1037 x = self.__findaddr(pkt)
1038 return super(SourceIP6Field, self).i2h(pkt, x)
1041class DestIP6Field(IP6Field, DestField):
1042 bindings = {} # type: Dict[Type[Packet], Tuple[str, Any]]
1044 def __init__(self, name, default):
1045 # type: (str, str) -> None
1046 IP6Field.__init__(self, name, None)
1047 DestField.__init__(self, name, default)
1049 def i2m(self, pkt, x):
1050 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> bytes
1051 if x is None and pkt is not None:
1052 x = self.dst_from_pkt(pkt)
1053 return IP6Field.i2m(self, pkt, x)
1055 def i2h(self, pkt, x):
1056 # type: (Optional[Packet], Optional[Union[str, Net6]]) -> str
1057 if x is None and pkt is not None:
1058 x = self.dst_from_pkt(pkt)
1059 return super(DestIP6Field, self).i2h(pkt, x)
1062class ByteField(Field[int, int]):
1063 def __init__(self, name, default):
1064 # type: (str, Optional[int]) -> None
1065 Field.__init__(self, name, default, "B")
1068class XByteField(ByteField):
1069 def i2repr(self, pkt, x):
1070 # type: (Optional[Packet], int) -> str
1071 return lhex(self.i2h(pkt, x))
1074# XXX Unused field: at least add some tests
1075class OByteField(ByteField):
1076 def i2repr(self, pkt, x):
1077 # type: (Optional[Packet], int) -> str
1078 return "%03o" % self.i2h(pkt, x)
1081class ThreeBytesField(Field[int, int]):
1082 def __init__(self, name, default):
1083 # type: (str, int) -> None
1084 Field.__init__(self, name, default, "!I")
1085 self.sz = 3 # emits/consumes 3 bytes; keep i2len consistent for FieldLenField
1087 def addfield(self, pkt, s, val):
1088 # type: (Packet, bytes, Optional[int]) -> bytes
1089 return s + struct.pack(self.fmt, self.i2m(pkt, val))[1:4]
1091 def getfield(self, pkt, s):
1092 # type: (Packet, bytes) -> Tuple[bytes, int]
1093 return s[3:], self.m2i(pkt, struct.unpack(self.fmt, b"\x00" + s[:3])[0]) # noqa: E501
1096class X3BytesField(ThreeBytesField, XByteField):
1097 def i2repr(self, pkt, x):
1098 # type: (Optional[Packet], int) -> str
1099 return XByteField.i2repr(self, pkt, x)
1102class LEThreeBytesField(ByteField):
1103 def __init__(self, name, default):
1104 # type: (str, Optional[int]) -> None
1105 Field.__init__(self, name, default, "<I")
1106 self.sz = 3 # emits/consumes 3 bytes; keep i2len consistent for FieldLenField
1108 def addfield(self, pkt, s, val):
1109 # type: (Packet, bytes, Optional[int]) -> bytes
1110 return s + struct.pack(self.fmt, self.i2m(pkt, val))[:3]
1112 def getfield(self, pkt, s):
1113 # type: (Optional[Packet], bytes) -> Tuple[bytes, int]
1114 return s[3:], self.m2i(pkt, struct.unpack(self.fmt, s[:3] + b"\x00")[0]) # noqa: E501
1117class XLE3BytesField(LEThreeBytesField, XByteField):
1118 def i2repr(self, pkt, x):
1119 # type: (Optional[Packet], int) -> str
1120 return XByteField.i2repr(self, pkt, x)
1123def LEX3BytesField(*args, **kwargs):
1124 # type: (*Any, **Any) -> Any
1125 warnings.warn(
1126 "LEX3BytesField is deprecated. Use XLE3BytesField",
1127 DeprecationWarning
1128 )
1129 return XLE3BytesField(*args, **kwargs)
1132class NBytesField(Field[int, List[int]]):
1133 def __init__(self, name, default, sz):
1134 # type: (str, Optional[int], int) -> None
1135 Field.__init__(self, name, default, "<" + "B" * sz)
1137 def i2m(self, pkt, x):
1138 # type: (Optional[Packet], Optional[int]) -> List[int]
1139 if x is None:
1140 return [0] * self.sz
1141 x2m = list()
1142 for _ in range(self.sz):
1143 x2m.append(x % 256)
1144 x //= 256
1145 return x2m[::-1]
1147 def m2i(self, pkt, x):
1148 # type: (Optional[Packet], Union[List[int], int]) -> int
1149 if isinstance(x, int):
1150 return x
1151 # x can be a tuple when coming from struct.unpack (from getfield)
1152 if isinstance(x, (list, tuple)):
1153 return sum(d * (256 ** i) for i, d in enumerate(reversed(x)))
1154 return 0
1156 def i2repr(self, pkt, x):
1157 # type: (Optional[Packet], int) -> str
1158 if isinstance(x, int):
1159 return '%i' % x
1160 return super(NBytesField, self).i2repr(pkt, x)
1162 def addfield(self, pkt, s, val):
1163 # type: (Optional[Packet], bytes, Optional[int]) -> bytes
1164 return s + self.struct.pack(*self.i2m(pkt, val))
1166 def getfield(self, pkt, s):
1167 # type: (Optional[Packet], bytes) -> Tuple[bytes, int]
1168 return (s[self.sz:],
1169 self.m2i(pkt, self.struct.unpack(s[:self.sz]))) # type: ignore
1171 def randval(self):
1172 # type: () -> RandNum
1173 return RandNum(0, 2 ** (self.sz * 8) - 1)
1176class XNBytesField(NBytesField):
1177 def i2repr(self, pkt, x):
1178 # type: (Optional[Packet], int) -> str
1179 if isinstance(x, int):
1180 return '0x%x' % x
1181 # x can be a tuple when coming from struct.unpack (from getfield)
1182 if isinstance(x, (list, tuple)):
1183 return "0x" + "".join("%02x" % b for b in x)
1184 return super(XNBytesField, self).i2repr(pkt, x)
1187class SignedByteField(Field[int, int]):
1188 def __init__(self, name, default):
1189 # type: (str, Optional[int]) -> None
1190 Field.__init__(self, name, default, "b")
1193class FieldValueRangeException(Scapy_Exception):
1194 pass
1197class MaximumItemsCount(Scapy_Exception):
1198 pass
1201class FieldAttributeException(Scapy_Exception):
1202 pass
1205class YesNoByteField(ByteField):
1206 """
1207 A byte based flag field that shows representation of its number
1208 based on a given association
1210 In its default configuration the following representation is generated:
1211 x == 0 : 'no'
1212 x != 0 : 'yes'
1214 In more sophisticated use-cases (e.g. yes/no/invalid) one can use the
1215 config attribute to configure.
1216 Key-value, key-range and key-value-set associations that will be used to
1217 generate the values representation.
1219 - A range is given by a tuple (<first-val>, <last-value>) including the
1220 last value.
1221 - A single-value tuple is treated as scalar.
1222 - A list defines a set of (probably non consecutive) values that should be
1223 associated to a given key.
1225 All values not associated with a key will be shown as number of type
1226 unsigned byte.
1228 **For instance**::
1230 config = {
1231 'no' : 0,
1232 'foo' : (1,22),
1233 'yes' : 23,
1234 'bar' : [24,25, 42, 48, 87, 253]
1235 }
1237 Generates the following representations::
1239 x == 0 : 'no'
1240 x == 15: 'foo'
1241 x == 23: 'yes'
1242 x == 42: 'bar'
1243 x == 43: 43
1245 Another example, using the config attribute one could also revert
1246 the stock-yes-no-behavior::
1248 config = {
1249 'yes' : 0,
1250 'no' : (1,255)
1251 }
1253 Will generate the following value representation::
1255 x == 0 : 'yes'
1256 x != 0 : 'no'
1258 """
1259 __slots__ = ['eval_fn']
1261 def _build_config_representation(self, config):
1262 # type: (Dict[str, Any]) -> None
1263 assoc_table = dict()
1264 for key in config:
1265 value_spec = config[key]
1267 value_spec_type = type(value_spec)
1269 if value_spec_type is int:
1270 if value_spec < 0 or value_spec > 255:
1271 raise FieldValueRangeException('given field value {} invalid - ' # noqa: E501
1272 'must be in range [0..255]'.format(value_spec)) # noqa: E501
1273 assoc_table[value_spec] = key
1275 elif value_spec_type is list:
1276 for value in value_spec:
1277 if value < 0 or value > 255:
1278 raise FieldValueRangeException('given field value {} invalid - ' # noqa: E501
1279 'must be in range [0..255]'.format(value)) # noqa: E501
1280 assoc_table[value] = key
1282 elif value_spec_type is tuple:
1283 value_spec_len = len(value_spec)
1284 if value_spec_len != 2:
1285 raise FieldAttributeException('invalid length {} of given config item tuple {} - must be ' # noqa: E501
1286 '(<start-range>, <end-range>).'.format(value_spec_len, value_spec)) # noqa: E501
1288 value_range_start = value_spec[0]
1289 if value_range_start < 0 or value_range_start > 255:
1290 raise FieldValueRangeException('given field value {} invalid - ' # noqa: E501
1291 'must be in range [0..255]'.format(value_range_start)) # noqa: E501
1293 value_range_end = value_spec[1]
1294 if value_range_end < 0 or value_range_end > 255:
1295 raise FieldValueRangeException('given field value {} invalid - ' # noqa: E501
1296 'must be in range [0..255]'.format(value_range_end)) # noqa: E501
1298 for value in range(value_range_start, value_range_end + 1):
1300 assoc_table[value] = key
1302 self.eval_fn = lambda x: assoc_table[x] if x in assoc_table else x
1304 def __init__(self, name, default, config=None):
1305 # type: (str, int, Optional[Dict[str, Any]]) -> None
1307 if not config:
1308 # this represents the common use case and therefore it is kept small # noqa: E501
1309 self.eval_fn = lambda x: 'no' if x == 0 else 'yes'
1310 else:
1311 self._build_config_representation(config)
1312 ByteField.__init__(self, name, default)
1314 def i2repr(self, pkt, x):
1315 # type: (Optional[Packet], int) -> str
1316 return self.eval_fn(x) # type: ignore
1319class ShortField(Field[int, int]):
1320 def __init__(self, name, default):
1321 # type: (str, Optional[int]) -> None
1322 Field.__init__(self, name, default, "H")
1325class SignedShortField(Field[int, int]):
1326 def __init__(self, name, default):
1327 # type: (str, Optional[int]) -> None
1328 Field.__init__(self, name, default, "h")
1331class LEShortField(Field[int, int]):
1332 def __init__(self, name, default):
1333 # type: (str, Optional[int]) -> None
1334 Field.__init__(self, name, default, "<H")
1337class LESignedShortField(Field[int, int]):
1338 def __init__(self, name, default):
1339 # type: (str, Optional[int]) -> None
1340 Field.__init__(self, name, default, "<h")
1343class XShortField(ShortField):
1344 def i2repr(self, pkt, x):
1345 # type: (Optional[Packet], int) -> str
1346 return lhex(self.i2h(pkt, x))
1349class IntField(Field[int, int]):
1350 def __init__(self, name, default):
1351 # type: (str, Optional[int]) -> None
1352 Field.__init__(self, name, default, "I")
1355class SignedIntField(Field[int, int]):
1356 def __init__(self, name, default):
1357 # type: (str, int) -> None
1358 Field.__init__(self, name, default, "i")
1361class LEIntField(Field[int, int]):
1362 def __init__(self, name, default):
1363 # type: (str, Optional[int]) -> None
1364 Field.__init__(self, name, default, "<I")
1367class LESignedIntField(Field[int, int]):
1368 def __init__(self, name, default):
1369 # type: (str, int) -> None
1370 Field.__init__(self, name, default, "<i")
1373class XIntField(IntField):
1374 def i2repr(self, pkt, x):
1375 # type: (Optional[Packet], int) -> str
1376 return lhex(self.i2h(pkt, x))
1379class XLEIntField(LEIntField, XIntField):
1380 def i2repr(self, pkt, x):
1381 # type: (Optional[Packet], int) -> str
1382 return XIntField.i2repr(self, pkt, x)
1385class XLEShortField(LEShortField, XShortField):
1386 def i2repr(self, pkt, x):
1387 # type: (Optional[Packet], int) -> str
1388 return XShortField.i2repr(self, pkt, x)
1391class LongField(Field[int, int]):
1392 def __init__(self, name, default):
1393 # type: (str, int) -> None
1394 Field.__init__(self, name, default, "Q")
1397class SignedLongField(Field[int, int]):
1398 def __init__(self, name, default):
1399 # type: (str, Optional[int]) -> None
1400 Field.__init__(self, name, default, "q")
1403class LELongField(LongField):
1404 def __init__(self, name, default):
1405 # type: (str, Optional[int]) -> None
1406 Field.__init__(self, name, default, "<Q")
1409class LESignedLongField(Field[int, int]):
1410 def __init__(self, name, default):
1411 # type: (str, Optional[Any]) -> None
1412 Field.__init__(self, name, default, "<q")
1415class XLongField(LongField):
1416 def i2repr(self, pkt, x):
1417 # type: (Optional[Packet], int) -> str
1418 return lhex(self.i2h(pkt, x))
1421class XLELongField(LELongField, XLongField):
1422 def i2repr(self, pkt, x):
1423 # type: (Optional[Packet], int) -> str
1424 return XLongField.i2repr(self, pkt, x)
1427class IEEEFloatField(Field[int, int]):
1428 def __init__(self, name, default):
1429 # type: (str, Optional[int]) -> None
1430 Field.__init__(self, name, default, "f")
1433class IEEEDoubleField(Field[int, int]):
1434 def __init__(self, name, default):
1435 # type: (str, Optional[int]) -> None
1436 Field.__init__(self, name, default, "d")
1439class _StrField(Field[I, bytes]):
1440 __slots__ = ["remain"]
1442 def __init__(self, name, default, fmt="H", remain=0):
1443 # type: (str, Optional[I], str, int) -> None
1444 Field.__init__(self, name, default, fmt)
1445 self.remain = remain
1447 def i2len(self, pkt, x):
1448 # type: (Optional[Packet], Any) -> int
1449 if x is None:
1450 return 0
1451 return len(x)
1453 def any2i(self, pkt, x):
1454 # type: (Optional[Packet], Any) -> I
1455 if isinstance(x, str):
1456 x = bytes_encode(x)
1457 return super(_StrField, self).any2i(pkt, x) # type: ignore
1459 def i2repr(self, pkt, x):
1460 # type: (Optional[Packet], I) -> str
1461 if x and isinstance(x, bytes):
1462 return repr(x)
1463 return super(_StrField, self).i2repr(pkt, x)
1465 def i2m(self, pkt, x):
1466 # type: (Optional[Packet], Optional[I]) -> bytes
1467 if x is None:
1468 return b""
1469 if not isinstance(x, bytes):
1470 return bytes_encode(x)
1471 return x
1473 def addfield(self, pkt, s, val):
1474 # type: (Packet, bytes, Optional[I]) -> bytes
1475 return s + self.i2m(pkt, val)
1477 def getfield(self, pkt, s):
1478 # type: (Packet, bytes) -> Tuple[bytes, I]
1479 if self.remain == 0:
1480 return b"", self.m2i(pkt, s)
1481 else:
1482 return s[-self.remain:], self.m2i(pkt, s[:-self.remain])
1484 def randval(self):
1485 # type: () -> RandBin
1486 return RandBin(RandNum(0, 1200))
1489class StrField(_StrField[bytes]):
1490 pass
1493class StrFieldUtf16(StrField):
1494 def any2i(self, pkt, x):
1495 # type: (Optional[Packet], Optional[str]) -> bytes
1496 if isinstance(x, str):
1497 return self.h2i(pkt, x)
1498 return super(StrFieldUtf16, self).any2i(pkt, x)
1500 def i2repr(self, pkt, x):
1501 # type: (Optional[Packet], bytes) -> str
1502 return plain_str(self.i2h(pkt, x))
1504 def h2i(self, pkt, x):
1505 # type: (Optional[Packet], Optional[str]) -> bytes
1506 return plain_str(x).encode('utf-16-le', errors="replace")
1508 def i2h(self, pkt, x):
1509 # type: (Optional[Packet], bytes) -> str
1510 return bytes_encode(x).decode('utf-16-le', errors="replace")
1513class _StrEnumField:
1514 def __init__(self, **kwargs):
1515 # type: (**Any) -> None
1516 self.enum = kwargs.pop("enum", {})
1518 def i2repr(self, pkt, v):
1519 # type: (Optional[Packet], bytes) -> str
1520 r = v.rstrip(b"\0")
1521 rr = repr(r)
1522 if self.enum:
1523 if v in self.enum:
1524 rr = "%s (%s)" % (rr, self.enum[v])
1525 elif r in self.enum:
1526 rr = "%s (%s)" % (rr, self.enum[r])
1527 return rr
1530class StrEnumField(_StrEnumField, StrField):
1531 __slots__ = ["enum"]
1533 def __init__(
1534 self,
1535 name, # type: str
1536 default, # type: bytes
1537 enum=None, # type: Optional[Dict[str, str]]
1538 **kwargs # type: Any
1539 ):
1540 # type: (...) -> None
1541 StrField.__init__(self, name, default, **kwargs) # type: ignore
1542 self.enum = enum
1545K = TypeVar('K', List[BasePacket], BasePacket, Optional[BasePacket])
1548class _PacketField(_StrField[K]):
1549 __slots__ = ["cls"]
1550 holds_packets = 1
1552 def __init__(self,
1553 name, # type: str
1554 default, # type: Optional[K]
1555 pkt_cls, # type: Union[Callable[[bytes], Packet], Type[Packet]] # noqa: E501
1556 ):
1557 # type: (...) -> None
1558 super(_PacketField, self).__init__(name, default)
1559 self.cls = pkt_cls
1561 def i2m(self,
1562 pkt, # type: Optional[Packet]
1563 i, # type: Any
1564 ):
1565 # type: (...) -> bytes
1566 if i is None:
1567 return b""
1568 return raw(i)
1570 def m2i(self, pkt, m): # type: ignore
1571 # type: (Optional[Packet], bytes) -> Packet
1572 try:
1573 # we want to set parent wherever possible
1574 return self.cls(m, _parent=pkt) # type: ignore
1575 except TypeError:
1576 return self.cls(m)
1579class _PacketFieldSingle(_PacketField[K]):
1580 def any2i(self, pkt, x):
1581 # type: (Optional[Packet], Any) -> K
1582 if x and pkt and hasattr(x, "add_parent"):
1583 cast("Packet", x).add_parent(pkt)
1584 return super(_PacketFieldSingle, self).any2i(pkt, x)
1586 def getfield(self,
1587 pkt, # type: Packet
1588 s, # type: bytes
1589 ):
1590 # type: (...) -> Tuple[bytes, K]
1591 i = self.m2i(pkt, s)
1592 remain = b""
1593 if conf.padding_layer in i:
1594 r = i[conf.padding_layer]
1595 del r.underlayer.payload
1596 remain = r.load
1597 return remain, i # type: ignore
1600class PacketField(_PacketFieldSingle[BasePacket]):
1601 def randval(self): # type: ignore
1602 # type: () -> Packet
1603 from scapy.packet import fuzz
1604 return fuzz(self.cls()) # type: ignore
1607class PacketLenField(_PacketFieldSingle[Optional[BasePacket]]):
1608 __slots__ = ["length_from"]
1610 def __init__(self,
1611 name, # type: str
1612 default, # type: Packet
1613 cls, # type: Union[Callable[[bytes], Packet], Type[Packet]] # noqa: E501
1614 length_from=None # type: Optional[Callable[[Packet], int]] # noqa: E501
1615 ):
1616 # type: (...) -> None
1617 super(PacketLenField, self).__init__(name, default, cls)
1618 self.length_from = length_from or (lambda x: 0)
1620 def getfield(self,
1621 pkt, # type: Packet
1622 s, # type: bytes
1623 ):
1624 # type: (...) -> Tuple[bytes, Optional[BasePacket]]
1625 len_pkt = self.length_from(pkt)
1626 i = None
1627 if len_pkt:
1628 try:
1629 i = self.m2i(pkt, s[:len_pkt])
1630 except Exception:
1631 if conf.debug_dissector:
1632 raise
1633 i = conf.raw_layer(load=s[:len_pkt])
1634 return s[len_pkt:], i
1637class PacketListField(_PacketField[List[BasePacket]]):
1638 """PacketListField represents a list containing a series of Packet instances
1639 that might occur right in the middle of another Packet field.
1640 This field type may also be used to indicate that a series of Packet
1641 instances have a sibling semantic instead of a parent/child relationship
1642 (i.e. a stack of layers). All elements in PacketListField have current
1643 packet referenced in parent field.
1644 """
1645 __slots__ = ["count_from", "length_from", "next_cls_cb", "max_count"]
1646 islist = 1
1648 def __init__(
1649 self,
1650 name, # type: str
1651 default, # type: Optional[List[BasePacket]]
1652 pkt_cls=None, # type: Optional[Union[Callable[[bytes], Packet], Type[Packet]]] # noqa: E501
1653 count_from=None, # type: Optional[Callable[[Packet], int]]
1654 length_from=None, # type: Optional[Callable[[Packet], int]]
1655 next_cls_cb=None, # type: Optional[Callable[[Packet, List[BasePacket], Optional[Packet], bytes], Optional[Type[Packet]]]] # noqa: E501
1656 max_count=None, # type: Optional[int]
1657 ):
1658 # type: (...) -> None
1659 """
1660 The number of Packet instances that are dissected by this field can
1661 be parametrized using one of three different mechanisms/parameters:
1663 * count_from: a callback that returns the number of Packet
1664 instances to dissect. The callback prototype is::
1666 count_from(pkt:Packet) -> int
1668 * length_from: a callback that returns the number of bytes that
1669 must be dissected by this field. The callback prototype is::
1671 length_from(pkt:Packet) -> int
1673 * next_cls_cb: a callback that enables a Scapy developer to
1674 dynamically discover if another Packet instance should be
1675 dissected or not. See below for this callback prototype.
1677 The bytes that are not consumed during the dissection of this field
1678 are passed to the next field of the current packet.
1680 For the serialization of such a field, the list of Packets that are
1681 contained in a PacketListField can be heterogeneous and is
1682 unrestricted.
1684 The type of the Packet instances that are dissected with this field is
1685 specified or discovered using one of the following mechanism:
1687 * the pkt_cls parameter may contain a callable that returns an
1688 instance of the dissected Packet. This may either be a
1689 reference of a Packet subclass (e.g. DNSRROPT in layers/dns.py)
1690 to generate an homogeneous PacketListField or a function
1691 deciding the type of the Packet instance
1692 (e.g. _CDPGuessAddrRecord in contrib/cdp.py)
1694 * the pkt_cls parameter may contain a class object with a defined
1695 ``dispatch_hook`` classmethod. That method must return a Packet
1696 instance. The ``dispatch_hook`` callmethod must implement the
1697 following prototype::
1699 dispatch_hook(cls,
1700 _pkt:Optional[Packet],
1701 *args, **kargs
1702 ) -> Type[Packet]
1704 The _pkt parameter may contain a reference to the packet
1705 instance containing the PacketListField that is being
1706 dissected.
1708 * the ``next_cls_cb`` parameter may contain a callable whose
1709 prototype is::
1711 cbk(pkt:Packet,
1712 lst:List[Packet],
1713 cur:Optional[Packet],
1714 remain:bytes,
1715 ) -> Optional[Type[Packet]]
1717 The pkt argument contains a reference to the Packet instance
1718 containing the PacketListField that is being dissected.
1719 The lst argument is the list of all Packet instances that were
1720 previously parsed during the current ``PacketListField``
1721 dissection, saved for the very last Packet instance.
1722 The cur argument contains a reference to that very last parsed
1723 ``Packet`` instance. The remain argument contains the bytes
1724 that may still be consumed by the current PacketListField
1725 dissection operation.
1727 This callback returns either the type of the next Packet to
1728 dissect or None to indicate that no more Packet are to be
1729 dissected.
1731 These four arguments allows a variety of dynamic discovery of
1732 the number of Packet to dissect and of the type of each one of
1733 these Packets, including: type determination based on current
1734 Packet instances or its underlayers, continuation based on the
1735 previously parsed Packet instances within that PacketListField,
1736 continuation based on a look-ahead on the bytes to be
1737 dissected...
1739 The pkt_cls and next_cls_cb parameters are semantically exclusive,
1740 although one could specify both. If both are specified, pkt_cls is
1741 silently ignored. The same is true for count_from and next_cls_cb.
1743 length_from and next_cls_cb are compatible and the dissection will
1744 end, whichever of the two stop conditions comes first.
1746 :param name: the name of the field
1747 :param default: the default value of this field; generally an empty
1748 Python list
1749 :param pkt_cls: either a callable returning a Packet instance or a
1750 class object defining a ``dispatch_hook`` class method
1751 :param count_from: a callback returning the number of Packet
1752 instances to dissect.
1753 :param length_from: a callback returning the number of bytes to dissect
1754 :param next_cls_cb: a callback returning either None or the type of
1755 the next Packet to dissect.
1756 :param max_count: an int containing the max amount of results. This is
1757 a safety mechanism, exceeding this value will raise a Scapy_Exception.
1758 """
1759 if default is None:
1760 default = [] # Create a new list for each instance
1761 super(PacketListField, self).__init__(
1762 name,
1763 default,
1764 pkt_cls # type: ignore
1765 )
1766 self.count_from = count_from
1767 self.length_from = length_from
1768 self.next_cls_cb = next_cls_cb
1769 self.max_count = max_count
1771 def any2i(self, pkt, x):
1772 # type: (Optional[Packet], Any) -> List[BasePacket]
1773 if not isinstance(x, list):
1774 if x and pkt and hasattr(x, "add_parent"):
1775 x.add_parent(pkt)
1776 return [x]
1777 elif pkt:
1778 for i in x:
1779 if not i or not hasattr(i, "add_parent"):
1780 continue
1781 i.add_parent(pkt)
1782 return x
1784 def i2count(self,
1785 pkt, # type: Optional[Packet]
1786 val, # type: List[BasePacket]
1787 ):
1788 # type: (...) -> int
1789 if isinstance(val, list):
1790 return len(val)
1791 return 1
1793 def i2len(self,
1794 pkt, # type: Optional[Packet]
1795 val, # type: List[Packet]
1796 ):
1797 # type: (...) -> int
1798 return sum(len(self.i2m(pkt, p)) for p in val)
1800 def getfield(self, pkt, s):
1801 # type: (Packet, bytes) -> Tuple[bytes, List[BasePacket]]
1802 c = len_pkt = cls = None
1803 if self.length_from is not None:
1804 len_pkt = self.length_from(pkt)
1805 elif self.count_from is not None:
1806 c = self.count_from(pkt)
1807 if self.next_cls_cb is not None:
1808 cls = self.next_cls_cb(pkt, [], None, s)
1809 c = 1
1810 if cls is None:
1811 c = 0
1813 lst = [] # type: List[BasePacket]
1814 ret = b""
1815 remain = s
1816 if len_pkt is not None:
1817 remain, ret = s[:len_pkt], s[len_pkt:]
1818 while remain:
1819 if c is not None:
1820 if c <= 0:
1821 break
1822 c -= 1
1823 try:
1824 if cls is not None:
1825 try:
1826 # we want to set parent wherever possible
1827 p = cls(remain, _parent=pkt)
1828 except TypeError:
1829 p = cls(remain)
1830 else:
1831 p = self.m2i(pkt, remain)
1832 except Exception:
1833 if conf.debug_dissector:
1834 raise
1835 p = conf.raw_layer(load=remain)
1836 remain = b""
1837 else:
1838 if conf.padding_layer in p:
1839 pad = p[conf.padding_layer]
1840 remain = pad.load
1841 del pad.underlayer.payload
1842 if self.next_cls_cb is not None:
1843 cls = self.next_cls_cb(pkt, lst, p, remain)
1844 if cls is not None:
1845 c = 0 if c is None else c
1846 c += 1
1847 else:
1848 remain = b""
1849 lst.append(p)
1850 if len(lst) > (self.max_count or conf.max_list_count):
1851 raise MaximumItemsCount(
1852 "Maximum amount of items reached in PacketListField: %s "
1853 "(defaults to conf.max_list_count)"
1854 % (self.max_count or conf.max_list_count)
1855 )
1857 if isinstance(remain, tuple):
1858 remain, nb = remain
1859 return (remain + ret, nb), lst
1860 else:
1861 return remain + ret, lst
1863 def i2m(self,
1864 pkt, # type: Optional[Packet]
1865 i, # type: Any
1866 ):
1867 # type: (...) -> bytes
1868 return bytes_encode(i)
1870 def addfield(self, pkt, s, val):
1871 # type: (Packet, bytes, Any) -> bytes
1872 return s + b"".join(self.i2m(pkt, v) for v in val)
1875class StrFixedLenField(StrField):
1876 __slots__ = ["length_from"]
1878 def __init__(
1879 self,
1880 name, # type: str
1881 default, # type: Optional[bytes]
1882 length=None, # type: Optional[int]
1883 length_from=None, # type: Optional[Callable[[Packet], int]]
1884 ):
1885 # type: (...) -> None
1886 super(StrFixedLenField, self).__init__(name, default)
1887 self.length_from = length_from or (lambda x: 0)
1888 if length is not None:
1889 self.sz = length
1890 self.length_from = lambda x, length=length: length # type: ignore
1892 def i2repr(self,
1893 pkt, # type: Optional[Packet]
1894 v, # type: bytes
1895 ):
1896 # type: (...) -> str
1897 if isinstance(v, bytes) and not conf.debug_strfixedlenfield:
1898 v = v.rstrip(b"\0")
1899 return super(StrFixedLenField, self).i2repr(pkt, v)
1901 def getfield(self, pkt, s):
1902 # type: (Packet, bytes) -> Tuple[bytes, bytes]
1903 len_pkt = self.length_from(pkt)
1904 if len_pkt == 0:
1905 return s, b""
1906 return s[len_pkt:], self.m2i(pkt, s[:len_pkt])
1908 def addfield(self, pkt, s, val):
1909 # type: (Packet, bytes, Optional[bytes]) -> bytes
1910 len_pkt = self.length_from(pkt)
1911 if len_pkt is None:
1912 return s + self.i2m(pkt, val)
1913 return s + struct.pack("%is" % len_pkt, self.i2m(pkt, val))
1915 def randval(self):
1916 # type: () -> RandBin
1917 try:
1918 return RandBin(self.length_from(None)) # type: ignore
1919 except Exception:
1920 return RandBin(RandNum(0, 200))
1923class StrFixedLenFieldUtf16(StrFixedLenField, StrFieldUtf16):
1924 pass
1927class StrFixedLenEnumField(_StrEnumField, StrFixedLenField):
1928 __slots__ = ["enum"]
1930 def __init__(
1931 self,
1932 name, # type: str
1933 default, # type: bytes
1934 enum=None, # type: Optional[Dict[str, str]]
1935 length=None, # type: Optional[int]
1936 length_from=None # type: Optional[Callable[[Optional[Packet]], int]] # noqa: E501
1937 ):
1938 # type: (...) -> None
1939 StrFixedLenField.__init__(self, name, default, length=length, length_from=length_from) # noqa: E501
1940 self.enum = enum
1943class NetBIOSNameField(StrFixedLenField):
1944 def __init__(self, name, default, length=31):
1945 # type: (str, bytes, int) -> None
1946 StrFixedLenField.__init__(self, name, default, length)
1948 def h2i(self, pkt, x):
1949 # type: (Optional[Packet], bytes) -> bytes
1950 if x and len(x) > 15:
1951 x = x[:15]
1952 return x
1954 def i2m(self, pkt, y):
1955 # type: (Optional[Packet], Optional[bytes]) -> bytes
1956 if pkt:
1957 len_pkt = self.length_from(pkt) // 2
1958 else:
1959 len_pkt = 0
1960 x = bytes_encode(y or b"") # type: bytes
1961 x += b" " * len_pkt
1962 x = x[:len_pkt]
1963 x = b"".join(
1964 struct.pack(
1965 "!BB",
1966 0x41 + (b >> 4),
1967 0x41 + (b & 0xf),
1968 )
1969 for b in x
1970 )
1971 return b" " + x
1973 def m2i(self, pkt, x):
1974 # type: (Optional[Packet], bytes) -> bytes
1975 x = x[1:].strip(b"\x00")
1976 return b"".join(map(
1977 lambda x, y: struct.pack(
1978 "!B",
1979 (((x - 1) & 0xf) << 4) + ((y - 1) & 0xf)
1980 ),
1981 x[::2], x[1::2]
1982 )).rstrip(b" ")
1985class StrLenField(StrField):
1986 """
1987 StrField with a length
1989 :param length_from: a function that returns the size of the string
1990 :param max_length: max size to use as randval
1991 """
1992 __slots__ = ["length_from", "max_length"]
1993 ON_WIRE_SIZE_UTF16 = True
1995 def __init__(
1996 self,
1997 name, # type: str
1998 default, # type: bytes
1999 length_from=None, # type: Optional[Callable[[Packet], int]]
2000 max_length=None, # type: Optional[Any]
2001 ):
2002 # type: (...) -> None
2003 super(StrLenField, self).__init__(name, default)
2004 self.length_from = length_from
2005 self.max_length = max_length
2007 def getfield(self, pkt, s):
2008 # type: (Any, bytes) -> Tuple[bytes, bytes]
2009 len_pkt = (self.length_from or (lambda x: 0))(pkt)
2010 if not self.ON_WIRE_SIZE_UTF16:
2011 len_pkt *= 2
2012 if len_pkt == 0:
2013 return s, b""
2014 return s[len_pkt:], self.m2i(pkt, s[:len_pkt])
2016 def randval(self):
2017 # type: () -> RandBin
2018 return RandBin(RandNum(0, self.max_length or 1200))
2021class _XStrField(Field[bytes, bytes]):
2022 def i2repr(self, pkt, x):
2023 # type: (Optional[Packet], bytes) -> str
2024 if isinstance(x, bytes):
2025 return bytes_hex(x).decode()
2026 return super(_XStrField, self).i2repr(pkt, x)
2029class XStrField(_XStrField, StrField):
2030 """
2031 StrField which value is printed as hexadecimal.
2032 """
2035class XStrLenField(_XStrField, StrLenField):
2036 """
2037 StrLenField which value is printed as hexadecimal.
2038 """
2041class XStrFixedLenField(_XStrField, StrFixedLenField):
2042 """
2043 StrFixedLenField which value is printed as hexadecimal.
2044 """
2047class XLEStrLenField(XStrLenField):
2048 def i2m(self, pkt, x):
2049 # type: (Optional[Packet], Optional[bytes]) -> bytes
2050 if not x:
2051 return b""
2052 return x[:: -1]
2054 def m2i(self, pkt, x):
2055 # type: (Optional[Packet], bytes) -> bytes
2056 return x[:: -1]
2059class StrLenFieldUtf16(StrLenField, StrFieldUtf16):
2060 pass
2063class StrLenEnumField(_StrEnumField, StrLenField):
2064 __slots__ = ["enum"]
2066 def __init__(
2067 self,
2068 name, # type: str
2069 default, # type: bytes
2070 enum=None, # type: Optional[Dict[str, str]]
2071 **kwargs # type: Any
2072 ):
2073 # type: (...) -> None
2074 StrLenField.__init__(self, name, default, **kwargs)
2075 self.enum = enum
2078class BoundStrLenField(StrLenField):
2079 __slots__ = ["minlen", "maxlen"]
2081 def __init__(
2082 self,
2083 name, # type: str
2084 default, # type: bytes
2085 minlen=0, # type: int
2086 maxlen=255, # type: int
2087 length_from=None # type: Optional[Callable[[Packet], int]]
2088 ):
2089 # type: (...) -> None
2090 StrLenField.__init__(self, name, default, length_from=length_from)
2091 self.minlen = minlen
2092 self.maxlen = maxlen
2094 def randval(self):
2095 # type: () -> RandBin
2096 return RandBin(RandNum(self.minlen, self.maxlen))
2099class FieldListField(Field[List[Any], List[Any]]):
2100 __slots__ = ["field", "count_from", "length_from", "max_count"]
2101 islist = 1
2103 def __init__(
2104 self,
2105 name, # type: str
2106 default, # type: Optional[List[AnyField]]
2107 field, # type: AnyField
2108 length_from=None, # type: Optional[Callable[[Packet], int]]
2109 count_from=None, # type: Optional[Callable[[Packet], int]]
2110 max_count=None, # type: Optional[int]
2111 ):
2112 # type: (...) -> None
2113 if default is None:
2114 default = [] # Create a new list for each instance
2115 self.field = field
2116 Field.__init__(self, name, default)
2117 self.count_from = count_from
2118 self.length_from = length_from
2119 self.max_count = max_count
2121 def i2count(self, pkt, val):
2122 # type: (Optional[Packet], List[Any]) -> int
2123 if isinstance(val, list):
2124 return len(val)
2125 return 1
2127 def i2len(self, pkt, val):
2128 # type: (Packet, List[Any]) -> int
2129 return int(sum(self.field.i2len(pkt, v) for v in val))
2131 def any2i(self, pkt, x):
2132 # type: (Optional[Packet], List[Any]) -> List[Any]
2133 if not isinstance(x, list):
2134 return [self.field.any2i(pkt, x)]
2135 else:
2136 return [self.field.any2i(pkt, e) for e in x]
2138 def i2repr(self,
2139 pkt, # type: Optional[Packet]
2140 x, # type: List[Any]
2141 ):
2142 # type: (...) -> str
2143 return "[%s]" % ", ".join(self.field.i2repr(pkt, v) for v in x)
2145 def addfield(self,
2146 pkt, # type: Packet
2147 s, # type: bytes
2148 val, # type: Optional[List[Any]]
2149 ):
2150 # type: (...) -> bytes
2151 val = self.i2m(pkt, val)
2152 for v in val:
2153 s = self.field.addfield(pkt, s, v)
2154 return s
2156 def getfield(self,
2157 pkt, # type: Packet
2158 s, # type: bytes
2159 ):
2160 # type: (...) -> Any
2161 c = len_pkt = None
2162 if self.length_from is not None:
2163 len_pkt = self.length_from(pkt)
2164 elif self.count_from is not None:
2165 c = self.count_from(pkt)
2167 val = []
2168 ret = b""
2169 if len_pkt is not None:
2170 s, ret = s[:len_pkt], s[len_pkt:]
2172 while s:
2173 if c is not None:
2174 if c <= 0:
2175 break
2176 c -= 1
2177 s, v = self.field.getfield(pkt, s)
2178 val.append(v)
2179 if len(val) > (self.max_count or conf.max_list_count):
2180 raise MaximumItemsCount(
2181 "Maximum amount of items reached in FieldListField: %s "
2182 "(defaults to conf.max_list_count)"
2183 % (self.max_count or conf.max_list_count)
2184 )
2186 if isinstance(s, tuple):
2187 s, bn = s
2188 return (s + ret, bn), val
2189 else:
2190 return s + ret, val
2193class FieldLenField(Field[int, int]):
2194 __slots__ = ["length_of", "count_of", "adjust"]
2196 def __init__(
2197 self,
2198 name, # type: str
2199 default, # type: Optional[Any]
2200 length_of=None, # type: Optional[str]
2201 fmt="H", # type: str
2202 count_of=None, # type: Optional[str]
2203 adjust=lambda pkt, x: x, # type: Callable[[Packet, int], int]
2204 ):
2205 # type: (...) -> None
2206 Field.__init__(self, name, default, fmt)
2207 self.length_of = length_of
2208 self.count_of = count_of
2209 self.adjust = adjust
2211 def i2m(self, pkt, x):
2212 # type: (Optional[Packet], Optional[int]) -> int
2213 if x is None and pkt is not None:
2214 if self.length_of is not None:
2215 fld, fval = pkt.getfield_and_val(self.length_of)
2216 f = fld.i2len(pkt, fval)
2217 elif self.count_of is not None:
2218 fld, fval = pkt.getfield_and_val(self.count_of)
2219 f = fld.i2count(pkt, fval)
2220 else:
2221 raise ValueError(
2222 "Field should have either length_of or count_of"
2223 )
2224 x = self.adjust(pkt, f)
2225 elif x is None:
2226 x = 0
2227 return x
2230class StrNullField(StrField):
2231 DELIMITER = b"\x00"
2233 def addfield(self, pkt, s, val):
2234 # type: (Packet, bytes, Optional[bytes]) -> bytes
2235 return s + self.i2m(pkt, val) + self.DELIMITER
2237 def getfield(self,
2238 pkt, # type: Packet
2239 s, # type: bytes
2240 ):
2241 # type: (...) -> Tuple[bytes, bytes]
2242 len_str = 0
2243 while True:
2244 len_str = s.find(self.DELIMITER, len_str)
2245 if len_str < 0:
2246 # DELIMITER not found: return empty
2247 return b"", s
2248 if len_str % len(self.DELIMITER):
2249 len_str += 1
2250 else:
2251 break
2252 return s[len_str + len(self.DELIMITER):], self.m2i(pkt, s[:len_str])
2254 def randval(self):
2255 # type: () -> RandTermString
2256 return RandTermString(RandNum(0, 1200), self.DELIMITER)
2258 def i2len(self, pkt, x):
2259 # type: (Optional[Packet], Any) -> int
2260 return super(StrNullField, self).i2len(pkt, x) + 1
2263class StrNullFieldUtf16(StrNullField, StrFieldUtf16):
2264 DELIMITER = b"\x00\x00"
2267class StrStopField(StrField):
2268 __slots__ = ["stop", "additional"]
2270 def __init__(self, name, default, stop, additional=0):
2271 # type: (str, str, bytes, int) -> None
2272 Field.__init__(self, name, default)
2273 self.stop = stop
2274 self.additional = additional
2276 def getfield(self, pkt, s):
2277 # type: (Optional[Packet], bytes) -> Tuple[bytes, bytes]
2278 len_str = s.find(self.stop)
2279 if len_str < 0:
2280 return b"", s
2281 len_str += len(self.stop) + self.additional
2282 return s[len_str:], s[:len_str]
2284 def randval(self):
2285 # type: () -> RandTermString
2286 return RandTermString(RandNum(0, 1200), self.stop)
2289class LenField(Field[int, int]):
2290 """
2291 If None, will be filled with the size of the payload
2292 """
2293 __slots__ = ["adjust"]
2295 def __init__(self, name, default, fmt="H", adjust=lambda x: x):
2296 # type: (str, Optional[Any], str, Callable[[int], int]) -> None
2297 Field.__init__(self, name, default, fmt)
2298 self.adjust = adjust
2300 def i2m(self,
2301 pkt, # type: Optional[Packet]
2302 x, # type: Optional[int]
2303 ):
2304 # type: (...) -> int
2305 if x is None:
2306 x = 0
2307 if pkt is not None:
2308 x = self.adjust(len(pkt.payload))
2309 return x
2312class BCDFloatField(Field[float, int]):
2313 def i2m(self, pkt, x):
2314 # type: (Optional[Packet], Optional[float]) -> int
2315 if x is None:
2316 return 0
2317 return int(256 * x)
2319 def m2i(self, pkt, x):
2320 # type: (Optional[Packet], int) -> float
2321 return x / 256.0
2324class _BitField(Field[I, int]):
2325 """
2326 Field to handle bits.
2328 :param name: name of the field
2329 :param default: default value
2330 :param size: size (in bits). If negative, Low endian
2331 :param tot_size: size of the total group of bits (in bytes) the bitfield
2332 is in. If negative, Low endian.
2333 :param end_tot_size: same but for the BitField ending a group.
2335 Example - normal usage::
2337 0 1 2 3
2338 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
2339 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
2340 | A | B | C |
2341 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
2343 Fig. TestPacket
2345 class TestPacket(Packet):
2346 fields_desc = [
2347 BitField("a", 0, 14),
2348 BitField("b", 0, 16),
2349 BitField("c", 0, 2),
2350 ]
2352 Example - Low endian stored as 16 bits on the network::
2354 x x x x x x x x x x x x x x x x
2355 a [b] [ c ] [ a ]
2357 Will first get reversed during dissecion:
2359 x x x x x x x x x x x x x x x x
2360 [ a ] [b] [ c ]
2362 class TestPacket(Packet):
2363 fields_desc = [
2364 BitField("a", 0, 9, tot_size=-2),
2365 BitField("b", 0, 2),
2366 BitField("c", 0, 5, end_tot_size=-2)
2367 ]
2369 """
2370 __slots__ = ["rev", "size", "tot_size", "end_tot_size"]
2372 def __init__(self, name, default, size,
2373 tot_size=0, end_tot_size=0):
2374 # type: (str, Optional[I], int, int, int) -> None
2375 Field.__init__(self, name, default)
2376 if callable(size):
2377 size = size(self)
2378 self.rev = size < 0 or tot_size < 0 or end_tot_size < 0
2379 self.size = abs(size)
2380 if not tot_size:
2381 tot_size = self.size // 8
2382 self.tot_size = abs(tot_size)
2383 if not end_tot_size:
2384 end_tot_size = self.size // 8
2385 self.end_tot_size = abs(end_tot_size)
2386 # Fields always have a round sz except BitField
2387 # so to keep it simple, we'll ignore it here.
2388 self.sz = self.size / 8. # type: ignore
2390 # We need to # type: ignore a few things because of how special
2391 # BitField is
2392 def addfield(self, # type: ignore
2393 pkt, # type: Packet
2394 s, # type: Union[Tuple[bytes, int, int], bytes]
2395 ival, # type: I
2396 ):
2397 # type: (...) -> Union[Tuple[bytes, int, int], bytes]
2398 val = self.i2m(pkt, ival)
2399 if isinstance(s, tuple):
2400 s, bitsdone, v = s
2401 else:
2402 bitsdone = 0
2403 v = 0
2404 v <<= self.size
2405 v |= val & ((1 << self.size) - 1)
2406 bitsdone += self.size
2407 while bitsdone >= 8:
2408 bitsdone -= 8
2409 s = s + struct.pack("!B", v >> bitsdone)
2410 v &= (1 << bitsdone) - 1
2411 if bitsdone:
2412 return s, bitsdone, v
2413 else:
2414 # Apply LE if necessary
2415 if self.rev and self.end_tot_size > 1:
2416 s = s[:-self.end_tot_size] + s[-self.end_tot_size:][::-1]
2417 return s
2419 def getfield(self, # type: ignore
2420 pkt, # type: Packet
2421 s, # type: Union[Tuple[bytes, int], bytes]
2422 ):
2423 # type: (...) -> Union[Tuple[Tuple[bytes, int], I], Tuple[bytes, I]] # noqa: E501
2424 if isinstance(s, tuple):
2425 s, bn = s
2426 else:
2427 bn = 0
2428 # Apply LE if necessary
2429 if self.rev and self.tot_size > 1:
2430 s = s[:self.tot_size][::-1] + s[self.tot_size:]
2432 # we don't want to process all the string
2433 nb_bytes = (self.size + bn - 1) // 8 + 1
2434 w = s[:nb_bytes]
2436 # split the substring byte by byte
2437 _bytes = struct.unpack('!%dB' % nb_bytes, w)
2439 b = 0
2440 for c in range(nb_bytes):
2441 b |= int(_bytes[c]) << (nb_bytes - c - 1) * 8
2443 # get rid of high order bits
2444 b &= (1 << (nb_bytes * 8 - bn)) - 1
2446 # remove low order bits
2447 b = b >> (nb_bytes * 8 - self.size - bn)
2449 bn += self.size
2450 s = s[bn // 8:]
2451 bn = bn % 8
2452 b2 = self.m2i(pkt, b)
2453 if bn:
2454 return (s, bn), b2
2455 else:
2456 return s, b2
2458 def randval(self):
2459 # type: () -> RandNum
2460 return RandNum(0, 2**self.size - 1)
2462 def i2len(self, pkt, x): # type: ignore
2463 # type: (Optional[Packet], Optional[float]) -> float
2464 return float(self.size) / 8
2467class BitField(_BitField[int]):
2468 __doc__ = _BitField.__doc__
2471class BitLenField(BitField):
2472 __slots__ = ["length_from"]
2474 def __init__(self,
2475 name, # type: str
2476 default, # type: Optional[int]
2477 length_from # type: Callable[[Packet], int]
2478 ):
2479 # type: (...) -> None
2480 self.length_from = length_from
2481 super(BitLenField, self).__init__(name, default, 0)
2483 def getfield(self, # type: ignore
2484 pkt, # type: Packet
2485 s, # type: Union[Tuple[bytes, int], bytes]
2486 ):
2487 # type: (...) -> Union[Tuple[Tuple[bytes, int], int], Tuple[bytes, int]] # noqa: E501
2488 self.size = self.length_from(pkt)
2489 return super(BitLenField, self).getfield(pkt, s)
2491 def addfield(self, # type: ignore
2492 pkt, # type: Packet
2493 s, # type: Union[Tuple[bytes, int, int], bytes]
2494 val # type: int
2495 ):
2496 # type: (...) -> Union[Tuple[bytes, int, int], bytes]
2497 self.size = self.length_from(pkt)
2498 return super(BitLenField, self).addfield(pkt, s, val)
2501class BitFieldLenField(BitField):
2502 __slots__ = ["length_of", "count_of", "adjust", "tot_size", "end_tot_size"]
2504 def __init__(self,
2505 name, # type: str
2506 default, # type: Optional[int]
2507 size, # type: int
2508 length_of=None, # type: Optional[Union[Callable[[Optional[Packet]], int], str]] # noqa: E501
2509 count_of=None, # type: Optional[str]
2510 adjust=lambda pkt, x: x, # type: Callable[[Optional[Packet], int], int] # noqa: E501
2511 tot_size=0, # type: int
2512 end_tot_size=0, # type: int
2513 ):
2514 # type: (...) -> None
2515 super(BitFieldLenField, self).__init__(name, default, size,
2516 tot_size, end_tot_size)
2517 self.length_of = length_of
2518 self.count_of = count_of
2519 self.adjust = adjust
2521 def i2m(self, pkt, x):
2522 # type: (Optional[Packet], Optional[Any]) -> int
2523 return FieldLenField.i2m(self, pkt, x) # type: ignore
2526class XBitField(BitField):
2527 def i2repr(self, pkt, x):
2528 # type: (Optional[Packet], int) -> str
2529 return lhex(self.i2h(pkt, x))
2532_EnumType = Union[Dict[I, str], Dict[str, I], List[str], DADict[I, str], Type[Enum], Tuple[Callable[[I], str], Callable[[str], I]]] # noqa: E501
2535class _EnumField(Field[Union[List[I], I], I]):
2536 def __init__(self,
2537 name, # type: str
2538 default, # type: Optional[I]
2539 enum, # type: _EnumType[I]
2540 fmt="H", # type: str
2541 ):
2542 # type: (...) -> None
2543 """ Initializes enum fields.
2545 @param name: name of this field
2546 @param default: default value of this field
2547 @param enum: either an enum, a dict or a tuple of two callables.
2548 Dict keys are the internal values, while the dict
2549 values are the user-friendly representations. If the
2550 tuple is provided, the first callable receives the
2551 internal value as parameter and returns the
2552 user-friendly representation and the second callable
2553 does the converse. The first callable may return None
2554 to default to a literal string (repr()) representation.
2555 @param fmt: struct.pack format used to parse and serialize the
2556 internal value from and to machine representation.
2557 """
2558 if isinstance(enum, ObservableDict):
2559 cast(ObservableDict, enum).observe(self)
2561 if isinstance(enum, tuple):
2562 self.i2s_cb = enum[0] # type: Optional[Callable[[I], str]]
2563 self.s2i_cb = enum[1] # type: Optional[Callable[[str], I]]
2564 self.i2s = None # type: Optional[Dict[I, str]]
2565 self.s2i = None # type: Optional[Dict[str, I]]
2566 elif isinstance(enum, type) and issubclass(enum, Enum):
2567 # Python's Enum
2568 i2s = self.i2s = {}
2569 s2i = self.s2i = {}
2570 self.i2s_cb = None
2571 self.s2i_cb = None
2572 names = [x.name for x in enum]
2573 for n in names:
2574 value = enum[n].value
2575 i2s[value] = n
2576 s2i[n] = value
2577 else:
2578 i2s = self.i2s = {}
2579 s2i = self.s2i = {}
2580 self.i2s_cb = None
2581 self.s2i_cb = None
2582 keys = [] # type: List[I]
2583 if isinstance(enum, list):
2584 keys = list(range(len(enum))) # type: ignore
2585 elif isinstance(enum, DADict):
2586 keys = enum.keys()
2587 else:
2588 keys = list(enum) # type: ignore
2589 if any(isinstance(x, str) for x in keys):
2590 i2s, s2i = s2i, i2s # type: ignore
2591 for k in keys:
2592 value = cast(str, enum[k]) # type: ignore
2593 i2s[k] = value
2594 s2i[value] = k
2595 Field.__init__(self, name, default, fmt)
2597 def any2i_one(self, pkt, x):
2598 # type: (Optional[Packet], Any) -> I
2599 if isinstance(x, Enum):
2600 return cast(I, x.value)
2601 elif isinstance(x, str):
2602 if self.s2i:
2603 x = self.s2i[x]
2604 elif self.s2i_cb:
2605 x = self.s2i_cb(x)
2606 return cast(I, x)
2608 def _i2repr(self, pkt, x):
2609 # type: (Optional[Packet], I) -> str
2610 return repr(x)
2612 def i2repr_one(self, pkt, x):
2613 # type: (Optional[Packet], I) -> str
2614 if self not in conf.noenum and not isinstance(x, VolatileValue):
2615 if self.i2s:
2616 try:
2617 return self.i2s[x]
2618 except KeyError:
2619 pass
2620 elif self.i2s_cb:
2621 ret = self.i2s_cb(x)
2622 if ret is not None:
2623 return ret
2624 return self._i2repr(pkt, x)
2626 def any2i(self, pkt, x):
2627 # type: (Optional[Packet], Any) -> Union[I, List[I]]
2628 if isinstance(x, list):
2629 return [self.any2i_one(pkt, z) for z in x]
2630 else:
2631 return self.any2i_one(pkt, x)
2633 def i2repr(self, pkt, x): # type: ignore
2634 # type: (Optional[Packet], Any) -> Union[List[str], str]
2635 if isinstance(x, list):
2636 return [self.i2repr_one(pkt, z) for z in x]
2637 else:
2638 return self.i2repr_one(pkt, x)
2640 def notify_set(self, enum, key, value):
2641 # type: (ObservableDict, I, str) -> None
2642 ks = "0x%x" if isinstance(key, int) else "%s"
2643 log_runtime.debug(
2644 "At %s: Change to %s at " + ks, self, value, key
2645 )
2646 if self.i2s is not None and self.s2i is not None:
2647 self.i2s[key] = value
2648 self.s2i[value] = key
2650 def notify_del(self, enum, key):
2651 # type: (ObservableDict, I) -> None
2652 ks = "0x%x" if isinstance(key, int) else "%s"
2653 log_runtime.debug("At %s: Delete value at " + ks, self, key)
2654 if self.i2s is not None and self.s2i is not None:
2655 value = self.i2s[key]
2656 del self.i2s[key]
2657 del self.s2i[value]
2660class EnumField(_EnumField[I]):
2661 __slots__ = ["i2s", "s2i", "s2i_cb", "i2s_cb"]
2664class CharEnumField(EnumField[str]):
2665 def __init__(self,
2666 name, # type: str
2667 default, # type: str
2668 enum, # type: _EnumType[str]
2669 fmt="1s", # type: str
2670 ):
2671 # type: (...) -> None
2672 super(CharEnumField, self).__init__(name, default, enum, fmt)
2673 if self.i2s is not None:
2674 k = list(self.i2s)
2675 if k and len(k[0]) != 1:
2676 self.i2s, self.s2i = self.s2i, self.i2s
2678 def any2i_one(self, pkt, x):
2679 # type: (Optional[Packet], str) -> str
2680 if len(x) != 1:
2681 if self.s2i:
2682 x = self.s2i[x]
2683 elif self.s2i_cb:
2684 x = self.s2i_cb(x)
2685 return x
2688class BitEnumField(_BitField[Union[List[int], int]], _EnumField[int]):
2689 __slots__ = EnumField.__slots__
2691 def __init__(self,
2692 name, # type: str
2693 default, # type: Optional[int]
2694 size, # type: int
2695 enum, # type: _EnumType[int]
2696 **kwargs # type: Any
2697 ):
2698 # type: (...) -> None
2699 _EnumField.__init__(self, name, default, enum)
2700 _BitField.__init__(self, name, default, size, **kwargs)
2702 def any2i(self, pkt, x):
2703 # type: (Optional[Packet], Any) -> Union[List[int], int]
2704 return _EnumField.any2i(self, pkt, x)
2706 def i2repr(self,
2707 pkt, # type: Optional[Packet]
2708 x, # type: Union[List[int], int]
2709 ):
2710 # type: (...) -> Any
2711 return _EnumField.i2repr(self, pkt, x)
2714class BitLenEnumField(BitLenField, _EnumField[int]):
2715 __slots__ = EnumField.__slots__
2717 def __init__(self,
2718 name, # type: str
2719 default, # type: Optional[int]
2720 length_from, # type: Callable[[Packet], int]
2721 enum, # type: _EnumType[int]
2722 **kwargs, # type: Any
2723 ):
2724 # type: (...) -> None
2725 _EnumField.__init__(self, name, default, enum)
2726 BitLenField.__init__(self, name, default, length_from, **kwargs)
2728 def any2i(self, pkt, x):
2729 # type: (Optional[Packet], Any) -> int
2730 return _EnumField.any2i(self, pkt, x) # type: ignore
2732 def i2repr(self,
2733 pkt, # type: Optional[Packet]
2734 x, # type: Union[List[int], int]
2735 ):
2736 # type: (...) -> Any
2737 return _EnumField.i2repr(self, pkt, x)
2740class ShortEnumField(EnumField[int]):
2741 __slots__ = EnumField.__slots__
2743 def __init__(self,
2744 name, # type: str
2745 default, # type: Optional[int]
2746 enum, # type: _EnumType[int]
2747 ):
2748 # type: (...) -> None
2749 super(ShortEnumField, self).__init__(name, default, enum, "H")
2752class LEShortEnumField(EnumField[int]):
2753 def __init__(self,
2754 name, # type: str
2755 default, # type: Optional[int]
2756 enum, # type: _EnumType[int]
2757 ):
2758 # type: (...) -> None
2759 super(LEShortEnumField, self).__init__(name, default, enum, "<H")
2762class LongEnumField(EnumField[int]):
2763 def __init__(self,
2764 name, # type: str
2765 default, # type: Optional[int]
2766 enum, # type: _EnumType[int]
2767 ):
2768 # type: (...) -> None
2769 super(LongEnumField, self).__init__(name, default, enum, "Q")
2772class LELongEnumField(EnumField[int]):
2773 def __init__(self,
2774 name, # type: str
2775 default, # type: Optional[int]
2776 enum, # type: _EnumType[int]
2777 ):
2778 # type: (...) -> None
2779 super(LELongEnumField, self).__init__(name, default, enum, "<Q")
2782class ByteEnumField(EnumField[int]):
2783 def __init__(self,
2784 name, # type: str
2785 default, # type: Optional[int]
2786 enum, # type: _EnumType[int]
2787 ):
2788 # type: (...) -> None
2789 super(ByteEnumField, self).__init__(name, default, enum, "B")
2792class XByteEnumField(ByteEnumField):
2793 def i2repr_one(self, pkt, x):
2794 # type: (Optional[Packet], int) -> str
2795 if self not in conf.noenum and not isinstance(x, VolatileValue):
2796 if self.i2s:
2797 try:
2798 return self.i2s[x]
2799 except KeyError:
2800 pass
2801 elif self.i2s_cb:
2802 ret = self.i2s_cb(x)
2803 if ret is not None:
2804 return ret
2805 return lhex(x)
2808class IntEnumField(EnumField[int]):
2809 def __init__(self,
2810 name, # type: str
2811 default, # type: Optional[int]
2812 enum, # type: _EnumType[int]
2813 ):
2814 # type: (...) -> None
2815 super(IntEnumField, self).__init__(name, default, enum, "I")
2818class SignedIntEnumField(EnumField[int]):
2819 def __init__(self,
2820 name, # type: str
2821 default, # type: Optional[int]
2822 enum, # type: _EnumType[int]
2823 ):
2824 # type: (...) -> None
2825 super(SignedIntEnumField, self).__init__(name, default, enum, "i")
2828class LEIntEnumField(EnumField[int]):
2829 def __init__(self,
2830 name, # type: str
2831 default, # type: Optional[int]
2832 enum, # type: _EnumType[int]
2833 ):
2834 # type: (...) -> None
2835 super(LEIntEnumField, self).__init__(name, default, enum, "<I")
2838class XLEIntEnumField(LEIntEnumField):
2839 def _i2repr(self, pkt, x):
2840 # type: (Optional[Packet], Any) -> str
2841 return lhex(x)
2844class XShortEnumField(ShortEnumField):
2845 def _i2repr(self, pkt, x):
2846 # type: (Optional[Packet], Any) -> str
2847 return lhex(x)
2850class LE3BytesEnumField(LEThreeBytesField, _EnumField[int]):
2851 __slots__ = EnumField.__slots__
2853 def __init__(self,
2854 name, # type: str
2855 default, # type: Optional[int]
2856 enum, # type: _EnumType[int]
2857 ):
2858 # type: (...) -> None
2859 _EnumField.__init__(self, name, default, enum)
2860 LEThreeBytesField.__init__(self, name, default)
2862 def any2i(self, pkt, x):
2863 # type: (Optional[Packet], Any) -> int
2864 return _EnumField.any2i(self, pkt, x) # type: ignore
2866 def i2repr(self, pkt, x): # type: ignore
2867 # type: (Optional[Packet], Any) -> Union[List[str], str]
2868 return _EnumField.i2repr(self, pkt, x)
2871class XLE3BytesEnumField(LE3BytesEnumField):
2872 def _i2repr(self, pkt, x):
2873 # type: (Optional[Packet], Any) -> str
2874 return lhex(x)
2877class _MultiEnumField(_EnumField[I]):
2878 def __init__(self,
2879 name, # type: str
2880 default, # type: int
2881 enum, # type: Dict[I, Dict[I, str]]
2882 depends_on, # type: Callable[[Optional[Packet]], I]
2883 fmt="H" # type: str
2884 ):
2885 # type: (...) -> None
2887 self.depends_on = depends_on
2888 self.i2s_multi = enum
2889 self.s2i_multi = {} # type: Dict[I, Dict[str, I]]
2890 self.s2i_all = {} # type: Dict[str, I]
2891 for m in enum:
2892 s2i = {} # type: Dict[str, I]
2893 self.s2i_multi[m] = s2i
2894 for k, v in enum[m].items():
2895 s2i[v] = k
2896 self.s2i_all[v] = k
2897 Field.__init__(self, name, default, fmt)
2899 def any2i_one(self, pkt, x):
2900 # type: (Optional[Packet], Any) -> I
2901 if isinstance(x, str):
2902 v = self.depends_on(pkt)
2903 if v in self.s2i_multi:
2904 s2i = self.s2i_multi[v]
2905 if x in s2i:
2906 return s2i[x]
2907 return self.s2i_all[x]
2908 return cast(I, x)
2910 def i2repr_one(self, pkt, x):
2911 # type: (Optional[Packet], I) -> str
2912 v = self.depends_on(pkt)
2913 if isinstance(v, VolatileValue):
2914 return repr(v)
2915 if v in self.i2s_multi:
2916 return str(self.i2s_multi[v].get(x, x))
2917 return str(x)
2920class MultiEnumField(_MultiEnumField[int], EnumField[int]):
2921 __slots__ = ["depends_on", "i2s_multi", "s2i_multi", "s2i_all"]
2924class BitMultiEnumField(_BitField[Union[List[int], int]],
2925 _MultiEnumField[int]):
2926 __slots__ = EnumField.__slots__ + MultiEnumField.__slots__
2928 def __init__(
2929 self,
2930 name, # type: str
2931 default, # type: int
2932 size, # type: int
2933 enum, # type: Dict[int, Dict[int, str]]
2934 depends_on # type: Callable[[Optional[Packet]], int]
2935 ):
2936 # type: (...) -> None
2937 _MultiEnumField.__init__(self, name, default, enum, depends_on)
2938 self.rev = size < 0
2939 self.size = abs(size)
2940 self.sz = self.size / 8. # type: ignore
2942 def any2i(self, pkt, x):
2943 # type: (Optional[Packet], Any) -> Union[List[int], int]
2944 return _MultiEnumField[int].any2i(
2945 self, # type: ignore
2946 pkt,
2947 x
2948 )
2950 def i2repr( # type: ignore
2951 self,
2952 pkt, # type: Optional[Packet]
2953 x # type: Union[List[int], int]
2954 ):
2955 # type: (...) -> Union[str, List[str]]
2956 return _MultiEnumField[int].i2repr(
2957 self, # type: ignore
2958 pkt,
2959 x
2960 )
2963class ByteEnumKeysField(ByteEnumField):
2964 """ByteEnumField that picks valid values when fuzzed. """
2966 def randval(self):
2967 # type: () -> RandEnumKeys
2968 return RandEnumKeys(self.i2s or {})
2971class ShortEnumKeysField(ShortEnumField):
2972 """ShortEnumField that picks valid values when fuzzed. """
2974 def randval(self):
2975 # type: () -> RandEnumKeys
2976 return RandEnumKeys(self.i2s or {})
2979class IntEnumKeysField(IntEnumField):
2980 """IntEnumField that picks valid values when fuzzed. """
2982 def randval(self):
2983 # type: () -> RandEnumKeys
2984 return RandEnumKeys(self.i2s or {})
2987# Little endian fixed length field
2990class LEFieldLenField(FieldLenField):
2991 def __init__(
2992 self,
2993 name, # type: str
2994 default, # type: Optional[Any]
2995 length_of=None, # type: Optional[str]
2996 fmt="<H", # type: str
2997 count_of=None, # type: Optional[str]
2998 adjust=lambda pkt, x: x, # type: Callable[[Packet, int], int]
2999 ):
3000 # type: (...) -> None
3001 FieldLenField.__init__(
3002 self, name, default,
3003 length_of=length_of,
3004 fmt=fmt,
3005 count_of=count_of,
3006 adjust=adjust
3007 )
3010class FlagValueIter(object):
3012 __slots__ = ["flagvalue", "cursor"]
3014 def __init__(self, flagvalue):
3015 # type: (FlagValue) -> None
3016 self.flagvalue = flagvalue
3017 self.cursor = 0
3019 def __iter__(self):
3020 # type: () -> FlagValueIter
3021 return self
3023 def __next__(self):
3024 # type: () -> str
3025 x = int(self.flagvalue)
3026 x >>= self.cursor
3027 while x:
3028 self.cursor += 1
3029 if x & 1:
3030 return self.flagvalue.names[self.cursor - 1]
3031 x >>= 1
3032 raise StopIteration
3034 next = __next__
3037class FlagValue(object):
3038 __slots__ = ["value", "names", "multi"]
3040 def _fixvalue(self, value):
3041 # type: (Any) -> int
3042 if not value:
3043 return 0
3044 if isinstance(value, str):
3045 value = value.split('+') if self.multi else list(value)
3046 if isinstance(value, list):
3047 y = 0
3048 for i in value:
3049 y |= 1 << self.names.index(i)
3050 value = y
3051 return int(value)
3053 def __init__(self, value, names):
3054 # type: (Union[List[str], int, str], Union[List[str], str]) -> None
3055 self.multi = isinstance(names, list)
3056 self.names = names
3057 self.value = self._fixvalue(value)
3059 def __hash__(self):
3060 # type: () -> int
3061 return hash(self.value)
3063 def __int__(self):
3064 # type: () -> int
3065 return self.value
3067 def __eq__(self, other):
3068 # type: (Any) -> bool
3069 return self.value == self._fixvalue(other)
3071 def __lt__(self, other):
3072 # type: (Any) -> bool
3073 return self.value < self._fixvalue(other)
3075 def __le__(self, other):
3076 # type: (Any) -> bool
3077 return self.value <= self._fixvalue(other)
3079 def __gt__(self, other):
3080 # type: (Any) -> bool
3081 return self.value > self._fixvalue(other)
3083 def __ge__(self, other):
3084 # type: (Any) -> bool
3085 return self.value >= self._fixvalue(other)
3087 def __ne__(self, other):
3088 # type: (Any) -> bool
3089 return self.value != self._fixvalue(other)
3091 def __and__(self, other):
3092 # type: (int) -> FlagValue
3093 return self.__class__(self.value & self._fixvalue(other), self.names)
3094 __rand__ = __and__
3096 def __or__(self, other):
3097 # type: (int) -> FlagValue
3098 return self.__class__(self.value | self._fixvalue(other), self.names)
3099 __ror__ = __or__
3100 __add__ = __or__ # + is an alias for |
3102 def __sub__(self, other):
3103 # type: (int) -> FlagValue
3104 return self.__class__(
3105 self.value & (2 ** len(self.names) - 1 - self._fixvalue(other)),
3106 self.names
3107 )
3109 def __xor__(self, other):
3110 # type: (int) -> FlagValue
3111 return self.__class__(self.value ^ self._fixvalue(other), self.names)
3113 def __lshift__(self, other):
3114 # type: (int) -> int
3115 return self.value << self._fixvalue(other)
3117 def __rshift__(self, other):
3118 # type: (int) -> int
3119 return self.value >> self._fixvalue(other)
3121 def __nonzero__(self):
3122 # type: () -> bool
3123 return bool(self.value)
3124 __bool__ = __nonzero__
3126 def flagrepr(self):
3127 # type: () -> str
3128 warnings.warn(
3129 "obj.flagrepr() is obsolete. Use str(obj) instead.",
3130 DeprecationWarning
3131 )
3132 return str(self)
3134 def __str__(self):
3135 # type: () -> str
3136 i = 0
3137 r = []
3138 x = int(self)
3139 while x:
3140 if x & 1:
3141 try:
3142 name = self.names[i]
3143 except IndexError:
3144 name = "?"
3145 r.append(name)
3146 i += 1
3147 x >>= 1
3148 return ("+" if self.multi else "").join(r)
3150 def __iter__(self):
3151 # type: () -> FlagValueIter
3152 return FlagValueIter(self)
3154 def __repr__(self):
3155 # type: () -> str
3156 return "<Flag %d (%s)>" % (self, self)
3158 def __deepcopy__(self, memo):
3159 # type: (Dict[Any, Any]) -> FlagValue
3160 return self.__class__(int(self), self.names)
3162 def __getattr__(self, attr):
3163 # type: (str) -> Any
3164 if attr in self.__slots__:
3165 return super(FlagValue, self).__getattribute__(attr)
3166 try:
3167 if self.multi:
3168 return bool((2 ** self.names.index(attr)) & int(self))
3169 return all(bool((2 ** self.names.index(flag)) & int(self))
3170 for flag in attr)
3171 except ValueError:
3172 if '_' in attr:
3173 try:
3174 return self.__getattr__(attr.replace('_', '-'))
3175 except AttributeError:
3176 pass
3177 return super(FlagValue, self).__getattribute__(attr)
3179 def __setattr__(self, attr, value):
3180 # type: (str, Union[List[str], int, str]) -> None
3181 if attr == "value" and not isinstance(value, int):
3182 raise ValueError(value)
3183 if attr in self.__slots__:
3184 return super(FlagValue, self).__setattr__(attr, value)
3185 if attr in self.names:
3186 if value:
3187 self.value |= (2 ** self.names.index(attr))
3188 else:
3189 self.value &= ~(2 ** self.names.index(attr))
3190 else:
3191 return super(FlagValue, self).__setattr__(attr, value)
3193 def copy(self):
3194 # type: () -> FlagValue
3195 return self.__class__(self.value, self.names)
3198class FlagsField(_BitField[Optional[Union[int, FlagValue]]]):
3199 """ Handle Flag type field
3201 Make sure all your flags have a label
3203 Example (list):
3204 >>> from scapy.packet import Packet
3205 >>> class FlagsTest(Packet):
3206 fields_desc = [FlagsField("flags", 0, 8, ["f0", "f1", "f2", "f3", "f4", "f5", "f6", "f7"])] # noqa: E501
3207 >>> FlagsTest(flags=9).show2()
3208 ###[ FlagsTest ]###
3209 flags = f0+f3
3211 Example (str):
3212 >>> from scapy.packet import Packet
3213 >>> class TCPTest(Packet):
3214 fields_desc = [
3215 BitField("reserved", 0, 7),
3216 FlagsField("flags", 0x2, 9, "FSRPAUECN")
3217 ]
3218 >>> TCPTest(flags=3).show2()
3219 ###[ FlagsTest ]###
3220 reserved = 0
3221 flags = FS
3223 Example (dict):
3224 >>> from scapy.packet import Packet
3225 >>> class FlagsTest2(Packet):
3226 fields_desc = [
3227 FlagsField("flags", 0x2, 16, {
3228 0x0001: "A",
3229 0x0008: "B",
3230 })
3231 ]
3233 :param name: field's name
3234 :param default: default value for the field
3235 :param size: number of bits in the field (in bits). if negative, LE
3236 :param names: (list or str or dict) label for each flag
3237 If it's a str or a list, the least Significant Bit tag's name
3238 is written first.
3239 """
3240 ismutable = True
3241 __slots__ = ["names"]
3243 def __init__(self,
3244 name, # type: str
3245 default, # type: Optional[Union[int, FlagValue]]
3246 size, # type: int
3247 names, # type: Union[List[str], str, Dict[int, str]]
3248 **kwargs # type: Any
3249 ):
3250 # type: (...) -> None
3251 # Convert the dict to a list
3252 if isinstance(names, dict):
3253 tmp = ["bit_%d" % i for i in range(abs(size))]
3254 for i, v in names.items():
3255 tmp[int(math.floor(math.log(i, 2)))] = v
3256 names = tmp
3257 # Store the names as str or list
3258 self.names = names
3259 super(FlagsField, self).__init__(name, default, size, **kwargs)
3261 def _fixup_val(self, x):
3262 # type: (Any) -> Optional[FlagValue]
3263 """Returns a FlagValue instance when needed. Internal method, to be
3264used in *2i() and i2*() methods.
3266 """
3267 if isinstance(x, (FlagValue, VolatileValue)):
3268 return x # type: ignore
3269 if x is None:
3270 return None
3271 return FlagValue(x, self.names)
3273 def any2i(self, pkt, x):
3274 # type: (Optional[Packet], Any) -> Optional[FlagValue]
3275 return self._fixup_val(super(FlagsField, self).any2i(pkt, x))
3277 def m2i(self, pkt, x):
3278 # type: (Optional[Packet], int) -> Optional[FlagValue]
3279 return self._fixup_val(super(FlagsField, self).m2i(pkt, x))
3281 def i2h(self, pkt, x):
3282 # type: (Optional[Packet], Any) -> Optional[FlagValue]
3283 return self._fixup_val(super(FlagsField, self).i2h(pkt, x))
3285 def i2repr(self,
3286 pkt, # type: Optional[Packet]
3287 x, # type: Any
3288 ):
3289 # type: (...) -> str
3290 if isinstance(x, (list, tuple)):
3291 return repr(type(x)(
3292 "None" if v is None else str(self._fixup_val(v)) for v in x
3293 ))
3294 return "None" if x is None else str(self._fixup_val(x))
3297MultiFlagsEntry = collections.namedtuple('MultiFlagsEntry', ['short', 'long'])
3300class MultiFlagsField(_BitField[Set[str]]):
3301 __slots__ = FlagsField.__slots__ + ["depends_on"]
3303 def __init__(self,
3304 name, # type: str
3305 default, # type: Set[str]
3306 size, # type: int
3307 names, # type: Dict[int, Dict[int, MultiFlagsEntry]]
3308 depends_on, # type: Callable[[Optional[Packet]], int]
3309 ):
3310 # type: (...) -> None
3311 self.names = names
3312 self.depends_on = depends_on
3313 super(MultiFlagsField, self).__init__(name, default, size)
3315 def any2i(self, pkt, x):
3316 # type: (Optional[Packet], Any) -> Set[str]
3317 if not isinstance(x, (set, int)):
3318 raise ValueError('set expected')
3320 if pkt is not None:
3321 if isinstance(x, int):
3322 return self.m2i(pkt, x)
3323 else:
3324 v = self.depends_on(pkt)
3325 if v is not None:
3326 assert v in self.names, 'invalid dependency'
3327 these_names = self.names[v]
3328 s = set()
3329 for i in x:
3330 for val in these_names.values():
3331 if val.short == i:
3332 s.add(i)
3333 break
3334 else:
3335 assert False, 'Unknown flag "{}" with this dependency'.format(i) # noqa: E501
3336 continue
3337 return s
3338 if isinstance(x, int):
3339 return set()
3340 return x
3342 def i2m(self, pkt, x):
3343 # type: (Optional[Packet], Optional[Set[str]]) -> int
3344 v = self.depends_on(pkt)
3345 these_names = self.names.get(v, {})
3347 r = 0
3348 if x is None:
3349 return r
3350 for flag_set in x:
3351 for i, val in these_names.items():
3352 if val.short == flag_set:
3353 r |= 1 << i
3354 break
3355 else:
3356 r |= 1 << int(flag_set[len('bit '):])
3357 return r
3359 def m2i(self, pkt, x):
3360 # type: (Optional[Packet], int) -> Set[str]
3361 v = self.depends_on(pkt)
3362 these_names = self.names.get(v, {})
3364 r = set()
3365 i = 0
3366 while x:
3367 if x & 1:
3368 if i in these_names:
3369 r.add(these_names[i].short)
3370 else:
3371 r.add('bit {}'.format(i))
3372 x >>= 1
3373 i += 1
3374 return r
3376 def i2repr(self, pkt, x):
3377 # type: (Optional[Packet], Set[str]) -> str
3378 v = self.depends_on(pkt)
3379 these_names = self.names.get(v, {})
3381 r = set()
3382 for flag_set in x:
3383 for i in these_names.values():
3384 if i.short == flag_set:
3385 r.add("{} ({})".format(i.long, i.short))
3386 break
3387 else:
3388 r.add(flag_set)
3389 return repr(r)
3392class FixedPointField(BitField):
3393 __slots__ = ['frac_bits']
3395 def __init__(self, name, default, size, frac_bits=16):
3396 # type: (str, int, int, int) -> None
3397 self.frac_bits = frac_bits
3398 super(FixedPointField, self).__init__(name, default, size)
3400 def any2i(self, pkt, val):
3401 # type: (Optional[Packet], Optional[float]) -> Optional[int]
3402 if val is None:
3403 return val
3404 ival = int(val)
3405 fract = int((val - ival) * 2**self.frac_bits)
3406 return (ival << self.frac_bits) | fract
3408 def i2h(self, pkt, val):
3409 # type: (Optional[Packet], Optional[int]) -> Optional[EDecimal]
3410 # A bit of trickery to get precise floats
3411 if val is None:
3412 return val
3413 int_part = val >> self.frac_bits
3414 pw = 2.0**self.frac_bits
3415 frac_part = EDecimal(val & (1 << self.frac_bits) - 1)
3416 frac_part /= pw # type: ignore
3417 return int_part + frac_part.normalize(int(math.log10(pw)))
3419 def i2repr(self, pkt, val):
3420 # type: (Optional[Packet], int) -> str
3421 return str(self.i2h(pkt, val))
3424# Base class for IPv4 and IPv6 Prefixes inspired by IPField and IP6Field.
3425# Machine values are encoded in a multiple of wordbytes bytes.
3426class _IPPrefixFieldBase(Field[Tuple[str, int], Tuple[bytes, int]]):
3427 __slots__ = ["wordbytes", "maxbytes", "aton", "ntoa", "length_from"]
3429 def __init__(
3430 self,
3431 name, # type: str
3432 default, # type: Tuple[str, int]
3433 wordbytes, # type: int
3434 maxbytes, # type: int
3435 aton, # type: Callable[..., Any]
3436 ntoa, # type: Callable[..., Any]
3437 length_from=None # type: Optional[Callable[[Packet], int]]
3438 ):
3439 # type: (...) -> None
3440 self.wordbytes = wordbytes
3441 self.maxbytes = maxbytes
3442 self.aton = aton
3443 self.ntoa = ntoa
3444 Field.__init__(self, name, default, "%is" % self.maxbytes)
3445 if length_from is None:
3446 length_from = lambda x: 0
3447 self.length_from = length_from
3449 def _numbytes(self, pfxlen):
3450 # type: (int) -> int
3451 wbits = self.wordbytes * 8
3452 return ((pfxlen + (wbits - 1)) // wbits) * self.wordbytes
3454 def h2i(self, pkt, x):
3455 # type: (Optional[Packet], str) -> Tuple[str, int]
3456 # "fc00:1::1/64" -> ("fc00:1::1", 64)
3457 [pfx, pfxlen] = x.split('/')
3458 self.aton(pfx) # check for validity
3459 return (pfx, int(pfxlen))
3461 def i2h(self, pkt, x):
3462 # type: (Optional[Packet], Tuple[str, int]) -> str
3463 # ("fc00:1::1", 64) -> "fc00:1::1/64"
3464 (pfx, pfxlen) = x
3465 return "%s/%i" % (pfx, pfxlen)
3467 def i2m(self,
3468 pkt, # type: Optional[Packet]
3469 x # type: Optional[Tuple[str, int]]
3470 ):
3471 # type: (...) -> Tuple[bytes, int]
3472 # ("fc00:1::1", 64) -> (b"\xfc\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01", 64) # noqa: E501
3473 if x is None:
3474 pfx, pfxlen = "", 0
3475 else:
3476 (pfx, pfxlen) = x
3477 s = self.aton(pfx)
3478 return (s[:self._numbytes(pfxlen)], pfxlen)
3480 def m2i(self, pkt, x):
3481 # type: (Optional[Packet], Tuple[bytes, int]) -> Tuple[str, int]
3482 # (b"\xfc\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01", 64) -> ("fc00:1::1", 64) # noqa: E501
3483 (s, pfxlen) = x
3485 if len(s) < self.maxbytes:
3486 s = s + (b"\0" * (self.maxbytes - len(s)))
3487 return (self.ntoa(s), pfxlen)
3489 def any2i(self, pkt, x):
3490 # type: (Optional[Packet], Optional[Any]) -> Tuple[str, int]
3491 if x is None:
3492 return (self.ntoa(b"\0" * self.maxbytes), 1)
3494 return self.h2i(pkt, x)
3496 def i2len(self, pkt, x):
3497 # type: (Packet, Tuple[str, int]) -> int
3498 (_, pfxlen) = x
3499 return pfxlen
3501 def addfield(self, pkt, s, val):
3502 # type: (Packet, bytes, Optional[Tuple[str, int]]) -> bytes
3503 (rawpfx, pfxlen) = self.i2m(pkt, val)
3504 fmt = "!%is" % self._numbytes(pfxlen)
3505 return s + struct.pack(fmt, rawpfx)
3507 def getfield(self, pkt, s):
3508 # type: (Packet, bytes) -> Tuple[bytes, Tuple[str, int]]
3509 pfxlen = self.length_from(pkt)
3510 numbytes = self._numbytes(pfxlen)
3511 fmt = "!%is" % numbytes
3512 return s[numbytes:], self.m2i(pkt, (struct.unpack(fmt, s[:numbytes])[0], pfxlen)) # noqa: E501
3515class IPPrefixField(_IPPrefixFieldBase):
3516 def __init__(
3517 self,
3518 name, # type: str
3519 default, # type: Tuple[str, int]
3520 wordbytes=1, # type: int
3521 length_from=None # type: Optional[Callable[[Packet], int]]
3522 ):
3523 _IPPrefixFieldBase.__init__(
3524 self,
3525 name,
3526 default,
3527 wordbytes,
3528 4,
3529 inet_aton,
3530 inet_ntoa,
3531 length_from
3532 )
3535class IP6PrefixField(_IPPrefixFieldBase):
3536 def __init__(
3537 self,
3538 name, # type: str
3539 default, # type: Tuple[str, int]
3540 wordbytes=1, # type: int
3541 length_from=None # type: Optional[Callable[[Packet], int]]
3542 ):
3543 # type: (...) -> None
3544 _IPPrefixFieldBase.__init__(
3545 self,
3546 name,
3547 default,
3548 wordbytes,
3549 16,
3550 lambda a: inet_pton(socket.AF_INET6, a),
3551 lambda n: inet_ntop(socket.AF_INET6, n),
3552 length_from
3553 )
3556class UTCTimeField(Field[float, int]):
3557 __slots__ = ["epoch", "delta", "strf",
3558 "use_msec", "use_micro", "use_nano", "custom_scaling"]
3560 def __init__(self,
3561 name, # type: str
3562 default, # type: int
3563 use_msec=False, # type: bool
3564 use_micro=False, # type: bool
3565 use_nano=False, # type: bool
3566 epoch=None, # type: Optional[Tuple[int, int, int, int, int, int, int, int, int]] # noqa: E501
3567 strf="%a, %d %b %Y %H:%M:%S %z", # type: str
3568 custom_scaling=None, # type: Optional[int]
3569 fmt="I" # type: str
3570 ):
3571 # type: (...) -> None
3572 Field.__init__(self, name, default, fmt=fmt)
3573 mk_epoch = EPOCH if epoch is None else calendar.timegm(epoch)
3574 self.epoch = mk_epoch
3575 self.delta = mk_epoch - EPOCH
3576 self.strf = strf
3577 self.use_msec = use_msec
3578 self.use_micro = use_micro
3579 self.use_nano = use_nano
3580 self.custom_scaling = custom_scaling
3582 def i2repr(self, pkt, x):
3583 # type: (Optional[Packet], float) -> str
3584 if x is None:
3585 x = time.time() - self.delta
3586 elif self.use_msec:
3587 x = x / 1e3
3588 elif self.use_micro:
3589 x = x / 1e6
3590 elif self.use_nano:
3591 x = x / 1e9
3592 elif self.custom_scaling:
3593 x = x / self.custom_scaling
3594 x += self.delta
3595 # To make negative timestamps work on all plateforms (e.g. Windows),
3596 # we need a trick.
3597 t = (
3598 datetime.datetime(1970, 1, 1) +
3599 datetime.timedelta(seconds=x)
3600 ).strftime(self.strf)
3601 return "%s (%d)" % (t, int(x))
3603 def i2m(self, pkt, x):
3604 # type: (Optional[Packet], Optional[float]) -> int
3605 if x is None:
3606 x = time.time() - self.delta
3607 if self.use_msec:
3608 x = x * 1e3
3609 elif self.use_micro:
3610 x = x * 1e6
3611 elif self.use_nano:
3612 x = x * 1e9
3613 elif self.custom_scaling:
3614 x = x * self.custom_scaling
3615 return int(x)
3616 return int(x)
3619class SecondsIntField(Field[float, int]):
3620 __slots__ = ["use_msec", "use_micro", "use_nano"]
3622 def __init__(self, name, default,
3623 use_msec=False,
3624 use_micro=False,
3625 use_nano=False):
3626 # type: (str, int, bool, bool, bool) -> None
3627 Field.__init__(self, name, default, "I")
3628 self.use_msec = use_msec
3629 self.use_micro = use_micro
3630 self.use_nano = use_nano
3632 def i2repr(self, pkt, x):
3633 # type: (Optional[Packet], Optional[float]) -> str
3634 if x is None:
3635 y = 0 # type: Union[int, float]
3636 elif self.use_msec:
3637 y = x / 1e3
3638 elif self.use_micro:
3639 y = x / 1e6
3640 elif self.use_nano:
3641 y = x / 1e9
3642 else:
3643 y = x
3644 return "%s sec" % y
3647class _ScalingField(object):
3648 def __init__(self,
3649 name, # type: str
3650 default, # type: float
3651 scaling=1, # type: Union[int, float]
3652 unit="", # type: str
3653 offset=0, # type: Union[int, float]
3654 ndigits=3, # type: int
3655 fmt="B", # type: str
3656 ):
3657 # type: (...) -> None
3658 self.scaling = scaling
3659 self.unit = unit
3660 self.offset = offset
3661 self.ndigits = ndigits
3662 Field.__init__(self, name, default, fmt) # type: ignore
3664 def i2m(self,
3665 pkt, # type: Optional[Packet]
3666 x # type: Optional[Union[int, float]]
3667 ):
3668 # type: (...) -> Union[int, float]
3669 if x is None:
3670 x = 0
3671 x = (x - self.offset) / self.scaling
3672 if isinstance(x, float) and self.fmt[-1] != "f": # type: ignore
3673 x = int(round(x))
3674 return x
3676 def m2i(self, pkt, x):
3677 # type: (Optional[Packet], Union[int, float]) -> Union[int, float]
3678 x = x * self.scaling + self.offset
3679 if isinstance(x, float) and self.fmt[-1] != "f": # type: ignore
3680 x = round(x, self.ndigits)
3681 return x
3683 def any2i(self, pkt, x):
3684 # type: (Optional[Packet], Any) -> Union[int, float, None]
3685 if isinstance(x, (str, bytes)):
3686 x = struct.unpack(self.fmt, bytes_encode(x))[0] # type: ignore
3687 x = self.m2i(pkt, x)
3688 if not isinstance(x, (int, float)) and x is not None:
3689 raise ValueError("Unknown type")
3690 return x
3692 def i2repr(self, pkt, x):
3693 # type: (Optional[Packet], Union[int, float]) -> str
3694 return "%s %s" % (
3695 self.i2h(pkt, x), # type: ignore
3696 self.unit
3697 )
3699 def randval(self):
3700 # type: () -> RandFloat
3701 value = Field.randval(self) # type: ignore
3702 if value is not None:
3703 min_val = round(value.min * self.scaling + self.offset,
3704 self.ndigits)
3705 max_val = round(value.max * self.scaling + self.offset,
3706 self.ndigits)
3708 return RandFloat(min(min_val, max_val), max(min_val, max_val))
3711class ScalingField(_ScalingField,
3712 Field[Union[int, float], Union[int, float]]):
3713 """ Handle physical values which are scaled and/or offset for communication
3715 Example:
3716 >>> from scapy.packet import Packet
3717 >>> class ScalingFieldTest(Packet):
3718 fields_desc = [ScalingField('data', 0, scaling=0.1, offset=-1, unit='mV')] # noqa: E501
3719 >>> ScalingFieldTest(data=10).show2()
3720 ###[ ScalingFieldTest ]###
3721 data= 10.0 mV
3722 >>> hexdump(ScalingFieldTest(data=10))
3723 0000 6E n
3724 >>> hexdump(ScalingFieldTest(data=b"\x6D"))
3725 0000 6D m
3726 >>> ScalingFieldTest(data=b"\x6D").show2()
3727 ###[ ScalingFieldTest ]###
3728 data= 9.9 mV
3730 bytes(ScalingFieldTest(...)) will produce 0x6E in this example.
3731 0x6E is 110 (decimal). This is calculated through the scaling factor
3732 and the offset. "data" was set to 10, which means, we want to transfer
3733 the physical value 10 mV. To calculate the value, which has to be
3734 sent on the bus, the offset has to subtracted and the scaling has to be
3735 applied by division through the scaling factor.
3736 bytes = (data - offset) / scaling
3737 bytes = ( 10 - (-1) ) / 0.1
3738 bytes = 110 = 0x6E
3740 If you want to force a certain internal value, you can assign a byte-
3741 string to the field (data=b"\x6D"). If a string of a bytes object is
3742 given to the field, no internal value conversion will be applied
3744 :param name: field's name
3745 :param default: default value for the field
3746 :param scaling: scaling factor for the internal value conversion
3747 :param unit: string for the unit representation of the internal value
3748 :param offset: value to offset the internal value during conversion
3749 :param ndigits: number of fractional digits for the internal conversion
3750 :param fmt: struct.pack format used to parse and serialize the internal value from and to machine representation # noqa: E501
3751 """
3754class BitScalingField(_ScalingField, BitField): # type: ignore
3755 """
3756 A ScalingField that is a BitField
3757 """
3759 def __init__(self, name, default, size, *args, **kwargs):
3760 # type: (str, int, int, *Any, **Any) -> None
3761 _ScalingField.__init__(self, name, default, *args, **kwargs)
3762 BitField.__init__(self, name, default, size) # type: ignore
3765class OUIField(X3BytesField):
3766 """
3767 A field designed to carry a OUI (3 bytes)
3768 """
3770 def i2repr(self, pkt, val):
3771 # type: (Optional[Packet], int) -> str
3772 by_val = struct.pack("!I", val or 0)[1:]
3773 oui = str2mac(by_val + b"\0" * 3)[:8]
3774 if conf.manufdb:
3775 fancy = conf.manufdb._get_manuf(oui)
3776 if fancy != oui:
3777 return "%s (%s)" % (fancy, oui)
3778 return oui
3781class UUIDField(Field[UUID, bytes]):
3782 """Field for UUID storage, wrapping Python's uuid.UUID type.
3784 The internal storage format of this field is ``uuid.UUID`` from the Python
3785 standard library.
3787 There are three formats (``uuid_fmt``) for this field type:
3789 * ``FORMAT_BE`` (default): the UUID is six fields in big-endian byte order,
3790 per RFC 4122.
3792 This format is used by DHCPv6 (RFC 6355) and most network protocols.
3794 * ``FORMAT_LE``: the UUID is six fields, with ``time_low``, ``time_mid``
3795 and ``time_high_version`` in little-endian byte order. This *doesn't*
3796 change the arrangement of the fields from RFC 4122.
3798 This format is used by Microsoft's COM/OLE libraries.
3800 * ``FORMAT_REV``: the UUID is a single 128-bit integer in little-endian
3801 byte order. This *changes the arrangement* of the fields.
3803 This format is used by Bluetooth Low Energy.
3805 Note: You should use the constants here.
3807 The "human encoding" of this field supports a number of different input
3808 formats, and wraps Python's ``uuid.UUID`` library appropriately:
3810 * Given a bytearray, bytes or str of 16 bytes, this class decodes UUIDs in
3811 wire format.
3813 * Given a bytearray, bytes or str of other lengths, this delegates to
3814 ``uuid.UUID`` the Python standard library. This supports a number of
3815 different encoding options -- see the Python standard library
3816 documentation for more details.
3818 * Given an int or long, presumed to be a 128-bit integer to pass to
3819 ``uuid.UUID``.
3821 * Given a tuple:
3823 * Tuples of 11 integers are treated as having the last 6 integers forming
3824 the ``node`` field, and are merged before being passed as a tuple of 6
3825 integers to ``uuid.UUID``.
3827 * Otherwise, the tuple is passed as the ``fields`` parameter to
3828 ``uuid.UUID`` directly without modification.
3830 ``uuid.UUID`` expects a tuple of 6 integers.
3832 Other types (such as ``uuid.UUID``) are passed through.
3833 """
3835 __slots__ = ["uuid_fmt"]
3837 FORMAT_BE = 0
3838 FORMAT_LE = 1
3839 FORMAT_REV = 2
3841 # Change this when we get new formats
3842 FORMATS = (FORMAT_BE, FORMAT_LE, FORMAT_REV)
3844 def __init__(self, name, default, uuid_fmt=FORMAT_BE):
3845 # type: (str, Optional[int], int) -> None
3846 self.uuid_fmt = uuid_fmt
3847 self._check_uuid_fmt()
3848 Field.__init__(self, name, default, "16s")
3850 def _check_uuid_fmt(self):
3851 # type: () -> None
3852 """Checks .uuid_fmt, and raises an exception if it is not valid."""
3853 if self.uuid_fmt not in UUIDField.FORMATS:
3854 raise FieldValueRangeException(
3855 "Unsupported uuid_fmt ({})".format(self.uuid_fmt))
3857 def i2m(self, pkt, x):
3858 # type: (Optional[Packet], Optional[UUID]) -> bytes
3859 self._check_uuid_fmt()
3860 if x is None:
3861 return b'\0' * 16
3862 if self.uuid_fmt == UUIDField.FORMAT_BE:
3863 return x.bytes
3864 elif self.uuid_fmt == UUIDField.FORMAT_LE:
3865 return x.bytes_le
3866 elif self.uuid_fmt == UUIDField.FORMAT_REV:
3867 return x.bytes[::-1]
3868 else:
3869 raise FieldAttributeException("Unknown fmt")
3871 def m2i(self,
3872 pkt, # type: Optional[Packet]
3873 x, # type: bytes
3874 ):
3875 # type: (...) -> UUID
3876 self._check_uuid_fmt()
3877 if self.uuid_fmt == UUIDField.FORMAT_BE:
3878 return UUID(bytes=x)
3879 elif self.uuid_fmt == UUIDField.FORMAT_LE:
3880 return UUID(bytes_le=x)
3881 elif self.uuid_fmt == UUIDField.FORMAT_REV:
3882 return UUID(bytes=x[::-1])
3883 else:
3884 raise FieldAttributeException("Unknown fmt")
3886 def any2i(self,
3887 pkt, # type: Optional[Packet]
3888 x # type: Any # noqa: E501
3889 ):
3890 # type: (...) -> Optional[UUID]
3891 # Python's uuid doesn't handle bytearray, so convert to an immutable
3892 # type first.
3893 if isinstance(x, bytearray):
3894 x = bytes_encode(x)
3896 if isinstance(x, int):
3897 u = UUID(int=x)
3898 elif isinstance(x, tuple):
3899 if len(x) == 11:
3900 # For compatibility with dce_rpc: this packs into a tuple where
3901 # elements 7..10 are the 48-bit node ID.
3902 node = 0
3903 for i in x[5:]:
3904 node = (node << 8) | i
3906 x = (x[0], x[1], x[2], x[3], x[4], node)
3908 u = UUID(fields=x)
3909 elif isinstance(x, (str, bytes)):
3910 if len(x) == 16:
3911 # Raw bytes
3912 u = self.m2i(pkt, bytes_encode(x))
3913 else:
3914 u = UUID(plain_str(x))
3915 elif isinstance(x, (UUID, RandUUID)):
3916 u = cast(UUID, x)
3917 else:
3918 return None
3919 return u
3921 @staticmethod
3922 def randval():
3923 # type: () -> RandUUID
3924 return RandUUID()
3927class UUIDEnumField(UUIDField, _EnumField[UUID]):
3928 __slots__ = EnumField.__slots__
3930 def __init__(self, name, default, enum, uuid_fmt=0):
3931 # type: (str, Optional[int], Any, int) -> None
3932 _EnumField.__init__(self, name, default, enum, "16s") # type: ignore
3933 UUIDField.__init__(self, name, default, uuid_fmt=uuid_fmt)
3935 def any2i(self, pkt, x):
3936 # type: (Optional[Packet], Any) -> UUID
3937 return _EnumField.any2i(self, pkt, x) # type: ignore
3939 def i2repr(self,
3940 pkt, # type: Optional[Packet]
3941 x, # type: UUID
3942 ):
3943 # type: (...) -> Any
3944 return _EnumField.i2repr(self, pkt, x)
3947class BitExtendedField(Field[Optional[int], int]):
3948 """
3949 Low E Bit Extended Field
3951 This type of field has a variable number of bytes. Each byte is defined
3952 as follows:
3953 - 7 bits of data
3954 - 1 bit an an extension bit:
3956 + 0 means it is last byte of the field ("stopping bit")
3957 + 1 means there is another byte after this one ("forwarding bit")
3959 To get the actual data, it is necessary to hop the binary data byte per
3960 byte and to check the extension bit until 0
3961 """
3963 __slots__ = ["extension_bit"]
3965 def __init__(self, name, default, extension_bit):
3966 # type: (str, Optional[Any], int) -> None
3967 Field.__init__(self, name, default, "B")
3968 assert extension_bit in [7, 0]
3969 self.extension_bit = extension_bit
3971 def addfield(self, pkt, s, val):
3972 # type: (Optional[Packet], bytes, Optional[int]) -> bytes
3973 val = self.i2m(pkt, val)
3974 if not val:
3975 return s + b"\0"
3976 rv = b""
3977 mask = 1 << self.extension_bit
3978 shift = (self.extension_bit + 1) % 8
3979 while val:
3980 bv = (val & 0x7F) << shift
3981 val = val >> 7
3982 if val:
3983 bv |= mask
3984 rv += struct.pack("!B", bv)
3985 return s + rv
3987 def getfield(self, pkt, s):
3988 # type: (Optional[Any], bytes) -> Tuple[bytes, Optional[int]]
3989 val = 0
3990 smask = 1 << self.extension_bit
3991 mask = 0xFF & ~ (1 << self.extension_bit)
3992 shift = (self.extension_bit + 1) % 8
3993 i = 0
3994 while s:
3995 val |= ((s[0] & mask) >> shift) << (7 * i)
3996 if (s[0] & smask) == 0: # extension bit is 0
3997 # end
3998 s = s[1:]
3999 break
4000 s = s[1:]
4001 i += 1
4002 return s, self.m2i(pkt, val)
4005class LSBExtendedField(BitExtendedField):
4006 # This is a BitExtendedField with the extension bit on LSB
4007 def __init__(self, name, default):
4008 # type: (str, Optional[Any]) -> None
4009 BitExtendedField.__init__(self, name, default, extension_bit=0)
4012class MSBExtendedField(BitExtendedField):
4013 # This is a BitExtendedField with the extension bit on MSB
4014 def __init__(self, name, default):
4015 # type: (str, Optional[Any]) -> None
4016 BitExtendedField.__init__(self, name, default, extension_bit=7)