Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/fields.py: 69%
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 # Dissection always appends Padding last, so checking the topmost
1594 # layer avoids a full recursive haslayer()+getlayer() lookup.
1595 r = i.lastlayer()
1596 if isinstance(r, conf.padding_layer):
1597 del r.underlayer.payload # type: ignore
1598 remain = r.load
1599 return remain, i # type: ignore
1602class PacketField(_PacketFieldSingle[BasePacket]):
1603 def randval(self): # type: ignore
1604 # type: () -> Packet
1605 from scapy.packet import fuzz
1606 return fuzz(self.cls()) # type: ignore
1609class PacketLenField(_PacketFieldSingle[Optional[BasePacket]]):
1610 __slots__ = ["length_from"]
1612 def __init__(self,
1613 name, # type: str
1614 default, # type: Packet
1615 cls, # type: Union[Callable[[bytes], Packet], Type[Packet]] # noqa: E501
1616 length_from=None # type: Optional[Callable[[Packet], int]] # noqa: E501
1617 ):
1618 # type: (...) -> None
1619 super(PacketLenField, self).__init__(name, default, cls)
1620 self.length_from = length_from or (lambda x: 0)
1622 def getfield(self,
1623 pkt, # type: Packet
1624 s, # type: bytes
1625 ):
1626 # type: (...) -> Tuple[bytes, Optional[BasePacket]]
1627 len_pkt = self.length_from(pkt)
1628 i = None
1629 if len_pkt:
1630 try:
1631 i = self.m2i(pkt, s[:len_pkt])
1632 except Exception:
1633 if conf.debug_dissector:
1634 raise
1635 i = conf.raw_layer(load=s[:len_pkt])
1636 return s[len_pkt:], i
1639class PacketListField(_PacketField[List[BasePacket]]):
1640 """PacketListField represents a list containing a series of Packet instances
1641 that might occur right in the middle of another Packet field.
1642 This field type may also be used to indicate that a series of Packet
1643 instances have a sibling semantic instead of a parent/child relationship
1644 (i.e. a stack of layers). All elements in PacketListField have current
1645 packet referenced in parent field.
1646 """
1647 __slots__ = ["count_from", "length_from", "next_cls_cb", "max_count"]
1648 islist = 1
1650 def __init__(
1651 self,
1652 name, # type: str
1653 default, # type: Optional[List[BasePacket]]
1654 pkt_cls=None, # type: Optional[Union[Callable[[bytes], Packet], Type[Packet]]] # noqa: E501
1655 count_from=None, # type: Optional[Callable[[Packet], int]]
1656 length_from=None, # type: Optional[Callable[[Packet], int]]
1657 next_cls_cb=None, # type: Optional[Callable[[Packet, List[BasePacket], Optional[Packet], bytes], Optional[Type[Packet]]]] # noqa: E501
1658 max_count=None, # type: Optional[int]
1659 ):
1660 # type: (...) -> None
1661 """
1662 The number of Packet instances that are dissected by this field can
1663 be parametrized using one of three different mechanisms/parameters:
1665 * count_from: a callback that returns the number of Packet
1666 instances to dissect. The callback prototype is::
1668 count_from(pkt:Packet) -> int
1670 * length_from: a callback that returns the number of bytes that
1671 must be dissected by this field. The callback prototype is::
1673 length_from(pkt:Packet) -> int
1675 * next_cls_cb: a callback that enables a Scapy developer to
1676 dynamically discover if another Packet instance should be
1677 dissected or not. See below for this callback prototype.
1679 The bytes that are not consumed during the dissection of this field
1680 are passed to the next field of the current packet.
1682 For the serialization of such a field, the list of Packets that are
1683 contained in a PacketListField can be heterogeneous and is
1684 unrestricted.
1686 The type of the Packet instances that are dissected with this field is
1687 specified or discovered using one of the following mechanism:
1689 * the pkt_cls parameter may contain a callable that returns an
1690 instance of the dissected Packet. This may either be a
1691 reference of a Packet subclass (e.g. DNSRROPT in layers/dns.py)
1692 to generate an homogeneous PacketListField or a function
1693 deciding the type of the Packet instance
1694 (e.g. _CDPGuessAddrRecord in contrib/cdp.py)
1696 * the pkt_cls parameter may contain a class object with a defined
1697 ``dispatch_hook`` classmethod. That method must return a Packet
1698 instance. The ``dispatch_hook`` callmethod must implement the
1699 following prototype::
1701 dispatch_hook(cls,
1702 _pkt:Optional[Packet],
1703 *args, **kargs
1704 ) -> Type[Packet]
1706 The _pkt parameter may contain a reference to the packet
1707 instance containing the PacketListField that is being
1708 dissected.
1710 * the ``next_cls_cb`` parameter may contain a callable whose
1711 prototype is::
1713 cbk(pkt:Packet,
1714 lst:List[Packet],
1715 cur:Optional[Packet],
1716 remain:bytes,
1717 ) -> Optional[Type[Packet]]
1719 The pkt argument contains a reference to the Packet instance
1720 containing the PacketListField that is being dissected.
1721 The lst argument is the list of all Packet instances that were
1722 previously parsed during the current ``PacketListField``
1723 dissection, saved for the very last Packet instance.
1724 The cur argument contains a reference to that very last parsed
1725 ``Packet`` instance. The remain argument contains the bytes
1726 that may still be consumed by the current PacketListField
1727 dissection operation.
1729 This callback returns either the type of the next Packet to
1730 dissect or None to indicate that no more Packet are to be
1731 dissected.
1733 These four arguments allows a variety of dynamic discovery of
1734 the number of Packet to dissect and of the type of each one of
1735 these Packets, including: type determination based on current
1736 Packet instances or its underlayers, continuation based on the
1737 previously parsed Packet instances within that PacketListField,
1738 continuation based on a look-ahead on the bytes to be
1739 dissected...
1741 The pkt_cls and next_cls_cb parameters are semantically exclusive,
1742 although one could specify both. If both are specified, pkt_cls is
1743 silently ignored. The same is true for count_from and next_cls_cb.
1745 length_from and next_cls_cb are compatible and the dissection will
1746 end, whichever of the two stop conditions comes first.
1748 :param name: the name of the field
1749 :param default: the default value of this field; generally an empty
1750 Python list
1751 :param pkt_cls: either a callable returning a Packet instance or a
1752 class object defining a ``dispatch_hook`` class method
1753 :param count_from: a callback returning the number of Packet
1754 instances to dissect.
1755 :param length_from: a callback returning the number of bytes to dissect
1756 :param next_cls_cb: a callback returning either None or the type of
1757 the next Packet to dissect.
1758 :param max_count: an int containing the max amount of results. This is
1759 a safety mechanism, exceeding this value will raise a Scapy_Exception.
1760 """
1761 if default is None:
1762 default = [] # Create a new list for each instance
1763 super(PacketListField, self).__init__(
1764 name,
1765 default,
1766 pkt_cls # type: ignore
1767 )
1768 self.count_from = count_from
1769 self.length_from = length_from
1770 self.next_cls_cb = next_cls_cb
1771 self.max_count = max_count
1773 def any2i(self, pkt, x):
1774 # type: (Optional[Packet], Any) -> List[BasePacket]
1775 if not isinstance(x, list):
1776 if x and pkt and hasattr(x, "add_parent"):
1777 x.add_parent(pkt)
1778 return [x]
1779 elif pkt:
1780 for i in x:
1781 if not i or not hasattr(i, "add_parent"):
1782 continue
1783 i.add_parent(pkt)
1784 return x
1786 def i2count(self,
1787 pkt, # type: Optional[Packet]
1788 val, # type: List[BasePacket]
1789 ):
1790 # type: (...) -> int
1791 if isinstance(val, list):
1792 return len(val)
1793 return 1
1795 def i2len(self,
1796 pkt, # type: Optional[Packet]
1797 val, # type: List[Packet]
1798 ):
1799 # type: (...) -> int
1800 return sum(len(self.i2m(pkt, p)) for p in val)
1802 def getfield(self, pkt, s):
1803 # type: (Packet, bytes) -> Tuple[bytes, List[BasePacket]]
1804 c = len_pkt = cls = None
1805 if self.length_from is not None:
1806 len_pkt = self.length_from(pkt)
1807 elif self.count_from is not None:
1808 c = self.count_from(pkt)
1809 if self.next_cls_cb is not None:
1810 cls = self.next_cls_cb(pkt, [], None, s)
1811 c = 1
1812 if cls is None:
1813 c = 0
1815 lst = [] # type: List[BasePacket]
1816 ret = b""
1817 remain = s
1818 if len_pkt is not None:
1819 remain, ret = s[:len_pkt], s[len_pkt:]
1820 while remain:
1821 if c is not None:
1822 if c <= 0:
1823 break
1824 c -= 1
1825 try:
1826 if cls is not None:
1827 try:
1828 # we want to set parent wherever possible
1829 p = cls(remain, _parent=pkt)
1830 except TypeError:
1831 p = cls(remain)
1832 else:
1833 p = self.m2i(pkt, remain)
1834 except Exception:
1835 if conf.debug_dissector:
1836 raise
1837 p = conf.raw_layer(load=remain)
1838 remain = b""
1839 else:
1840 pad = p.lastlayer()
1841 if isinstance(pad, conf.padding_layer):
1842 remain = pad.load
1843 del pad.underlayer.payload # type: ignore
1844 if self.next_cls_cb is not None:
1845 cls = self.next_cls_cb(pkt, lst, p, remain)
1846 if cls is not None:
1847 c = 0 if c is None else c
1848 c += 1
1849 else:
1850 remain = b""
1851 lst.append(p)
1852 if len(lst) > (self.max_count or conf.max_list_count):
1853 raise MaximumItemsCount(
1854 "Maximum amount of items reached in PacketListField: %s "
1855 "(defaults to conf.max_list_count)"
1856 % (self.max_count or conf.max_list_count)
1857 )
1859 if isinstance(remain, tuple):
1860 remain, nb = remain
1861 return (remain + ret, nb), lst
1862 else:
1863 return remain + ret, lst
1865 def i2m(self,
1866 pkt, # type: Optional[Packet]
1867 i, # type: Any
1868 ):
1869 # type: (...) -> bytes
1870 return bytes_encode(i)
1872 def addfield(self, pkt, s, val):
1873 # type: (Packet, bytes, Any) -> bytes
1874 return s + b"".join(self.i2m(pkt, v) for v in val)
1877class StrFixedLenField(StrField):
1878 __slots__ = ["length_from"]
1880 def __init__(
1881 self,
1882 name, # type: str
1883 default, # type: Optional[bytes]
1884 length=None, # type: Optional[int]
1885 length_from=None, # type: Optional[Callable[[Packet], int]]
1886 ):
1887 # type: (...) -> None
1888 super(StrFixedLenField, self).__init__(name, default)
1889 self.length_from = length_from or (lambda x: 0)
1890 if length is not None:
1891 self.sz = length
1892 self.length_from = lambda x, length=length: length # type: ignore
1894 def i2repr(self,
1895 pkt, # type: Optional[Packet]
1896 v, # type: bytes
1897 ):
1898 # type: (...) -> str
1899 if isinstance(v, bytes) and not conf.debug_strfixedlenfield:
1900 v = v.rstrip(b"\0")
1901 return super(StrFixedLenField, self).i2repr(pkt, v)
1903 def getfield(self, pkt, s):
1904 # type: (Packet, bytes) -> Tuple[bytes, bytes]
1905 len_pkt = self.length_from(pkt)
1906 if len_pkt == 0:
1907 return s, b""
1908 return s[len_pkt:], self.m2i(pkt, s[:len_pkt])
1910 def addfield(self, pkt, s, val):
1911 # type: (Packet, bytes, Optional[bytes]) -> bytes
1912 len_pkt = self.length_from(pkt)
1913 if len_pkt is None:
1914 return s + self.i2m(pkt, val)
1915 return s + struct.pack("%is" % len_pkt, self.i2m(pkt, val))
1917 def randval(self):
1918 # type: () -> RandBin
1919 try:
1920 return RandBin(self.length_from(None)) # type: ignore
1921 except Exception:
1922 return RandBin(RandNum(0, 200))
1925class StrFixedLenFieldUtf16(StrFixedLenField, StrFieldUtf16):
1926 pass
1929class StrFixedLenEnumField(_StrEnumField, StrFixedLenField):
1930 __slots__ = ["enum"]
1932 def __init__(
1933 self,
1934 name, # type: str
1935 default, # type: bytes
1936 enum=None, # type: Optional[Dict[str, str]]
1937 length=None, # type: Optional[int]
1938 length_from=None # type: Optional[Callable[[Optional[Packet]], int]] # noqa: E501
1939 ):
1940 # type: (...) -> None
1941 StrFixedLenField.__init__(self, name, default, length=length, length_from=length_from) # noqa: E501
1942 self.enum = enum
1945class NetBIOSNameField(StrFixedLenField):
1946 def __init__(self, name, default, length=31):
1947 # type: (str, bytes, int) -> None
1948 StrFixedLenField.__init__(self, name, default, length)
1950 def h2i(self, pkt, x):
1951 # type: (Optional[Packet], bytes) -> bytes
1952 if x and len(x) > 15:
1953 x = x[:15]
1954 return x
1956 def i2m(self, pkt, y):
1957 # type: (Optional[Packet], Optional[bytes]) -> bytes
1958 if pkt:
1959 len_pkt = self.length_from(pkt) // 2
1960 else:
1961 len_pkt = 0
1962 x = bytes_encode(y or b"") # type: bytes
1963 x += b" " * len_pkt
1964 x = x[:len_pkt]
1965 x = b"".join(
1966 struct.pack(
1967 "!BB",
1968 0x41 + (b >> 4),
1969 0x41 + (b & 0xf),
1970 )
1971 for b in x
1972 )
1973 return b" " + x
1975 def m2i(self, pkt, x):
1976 # type: (Optional[Packet], bytes) -> bytes
1977 x = x[1:].strip(b"\x00")
1978 return b"".join(map(
1979 lambda x, y: struct.pack(
1980 "!B",
1981 (((x - 1) & 0xf) << 4) + ((y - 1) & 0xf)
1982 ),
1983 x[::2], x[1::2]
1984 )).rstrip(b" ")
1987class StrLenField(StrField):
1988 """
1989 StrField with a length
1991 :param length_from: a function that returns the size of the string
1992 :param max_length: max size to use as randval
1993 """
1994 __slots__ = ["length_from", "max_length"]
1995 ON_WIRE_SIZE_UTF16 = True
1997 def __init__(
1998 self,
1999 name, # type: str
2000 default, # type: bytes
2001 length_from=None, # type: Optional[Callable[[Packet], int]]
2002 max_length=None, # type: Optional[Any]
2003 ):
2004 # type: (...) -> None
2005 super(StrLenField, self).__init__(name, default)
2006 self.length_from = length_from
2007 self.max_length = max_length
2009 def getfield(self, pkt, s):
2010 # type: (Any, bytes) -> Tuple[bytes, bytes]
2011 len_pkt = (self.length_from or (lambda x: 0))(pkt)
2012 if not self.ON_WIRE_SIZE_UTF16:
2013 len_pkt *= 2
2014 if len_pkt == 0:
2015 return s, b""
2016 return s[len_pkt:], self.m2i(pkt, s[:len_pkt])
2018 def randval(self):
2019 # type: () -> RandBin
2020 return RandBin(RandNum(0, self.max_length or 1200))
2023class _XStrField(Field[bytes, bytes]):
2024 def i2repr(self, pkt, x):
2025 # type: (Optional[Packet], bytes) -> str
2026 if isinstance(x, bytes):
2027 return bytes_hex(x).decode()
2028 return super(_XStrField, self).i2repr(pkt, x)
2031class XStrField(_XStrField, StrField):
2032 """
2033 StrField which value is printed as hexadecimal.
2034 """
2037class XStrLenField(_XStrField, StrLenField):
2038 """
2039 StrLenField which value is printed as hexadecimal.
2040 """
2043class XStrFixedLenField(_XStrField, StrFixedLenField):
2044 """
2045 StrFixedLenField which value is printed as hexadecimal.
2046 """
2049class XLEStrLenField(XStrLenField):
2050 def i2m(self, pkt, x):
2051 # type: (Optional[Packet], Optional[bytes]) -> bytes
2052 if not x:
2053 return b""
2054 return x[:: -1]
2056 def m2i(self, pkt, x):
2057 # type: (Optional[Packet], bytes) -> bytes
2058 return x[:: -1]
2061class StrLenFieldUtf16(StrLenField, StrFieldUtf16):
2062 pass
2065class StrLenEnumField(_StrEnumField, StrLenField):
2066 __slots__ = ["enum"]
2068 def __init__(
2069 self,
2070 name, # type: str
2071 default, # type: bytes
2072 enum=None, # type: Optional[Dict[str, str]]
2073 **kwargs # type: Any
2074 ):
2075 # type: (...) -> None
2076 StrLenField.__init__(self, name, default, **kwargs)
2077 self.enum = enum
2080class BoundStrLenField(StrLenField):
2081 __slots__ = ["minlen", "maxlen"]
2083 def __init__(
2084 self,
2085 name, # type: str
2086 default, # type: bytes
2087 minlen=0, # type: int
2088 maxlen=255, # type: int
2089 length_from=None # type: Optional[Callable[[Packet], int]]
2090 ):
2091 # type: (...) -> None
2092 StrLenField.__init__(self, name, default, length_from=length_from)
2093 self.minlen = minlen
2094 self.maxlen = maxlen
2096 def randval(self):
2097 # type: () -> RandBin
2098 return RandBin(RandNum(self.minlen, self.maxlen))
2101class FieldListField(Field[List[Any], List[Any]]):
2102 __slots__ = ["field", "count_from", "length_from", "max_count"]
2103 islist = 1
2105 def __init__(
2106 self,
2107 name, # type: str
2108 default, # type: Optional[List[AnyField]]
2109 field, # type: AnyField
2110 length_from=None, # type: Optional[Callable[[Packet], int]]
2111 count_from=None, # type: Optional[Callable[[Packet], int]]
2112 max_count=None, # type: Optional[int]
2113 ):
2114 # type: (...) -> None
2115 if default is None:
2116 default = [] # Create a new list for each instance
2117 self.field = field
2118 Field.__init__(self, name, default)
2119 self.count_from = count_from
2120 self.length_from = length_from
2121 self.max_count = max_count
2123 def i2count(self, pkt, val):
2124 # type: (Optional[Packet], List[Any]) -> int
2125 if isinstance(val, list):
2126 return len(val)
2127 return 1
2129 def i2len(self, pkt, val):
2130 # type: (Packet, List[Any]) -> int
2131 return int(sum(self.field.i2len(pkt, v) for v in val))
2133 def any2i(self, pkt, x):
2134 # type: (Optional[Packet], List[Any]) -> List[Any]
2135 if not isinstance(x, list):
2136 return [self.field.any2i(pkt, x)]
2137 else:
2138 return [self.field.any2i(pkt, e) for e in x]
2140 def i2repr(self,
2141 pkt, # type: Optional[Packet]
2142 x, # type: List[Any]
2143 ):
2144 # type: (...) -> str
2145 return "[%s]" % ", ".join(self.field.i2repr(pkt, v) for v in x)
2147 def addfield(self,
2148 pkt, # type: Packet
2149 s, # type: bytes
2150 val, # type: Optional[List[Any]]
2151 ):
2152 # type: (...) -> bytes
2153 val = self.i2m(pkt, val)
2154 for v in val:
2155 s = self.field.addfield(pkt, s, v)
2156 return s
2158 def getfield(self,
2159 pkt, # type: Packet
2160 s, # type: bytes
2161 ):
2162 # type: (...) -> Any
2163 c = len_pkt = None
2164 if self.length_from is not None:
2165 len_pkt = self.length_from(pkt)
2166 elif self.count_from is not None:
2167 c = self.count_from(pkt)
2169 val = []
2170 ret = b""
2171 if len_pkt is not None:
2172 s, ret = s[:len_pkt], s[len_pkt:]
2174 while s:
2175 if c is not None:
2176 if c <= 0:
2177 break
2178 c -= 1
2179 s, v = self.field.getfield(pkt, s)
2180 val.append(v)
2181 if len(val) > (self.max_count or conf.max_list_count):
2182 raise MaximumItemsCount(
2183 "Maximum amount of items reached in FieldListField: %s "
2184 "(defaults to conf.max_list_count)"
2185 % (self.max_count or conf.max_list_count)
2186 )
2188 if isinstance(s, tuple):
2189 s, bn = s
2190 return (s + ret, bn), val
2191 else:
2192 return s + ret, val
2195class FieldLenField(Field[int, int]):
2196 __slots__ = ["length_of", "count_of", "adjust"]
2198 def __init__(
2199 self,
2200 name, # type: str
2201 default, # type: Optional[Any]
2202 length_of=None, # type: Optional[str]
2203 fmt="H", # type: str
2204 count_of=None, # type: Optional[str]
2205 adjust=lambda pkt, x: x, # type: Callable[[Packet, int], int]
2206 ):
2207 # type: (...) -> None
2208 Field.__init__(self, name, default, fmt)
2209 self.length_of = length_of
2210 self.count_of = count_of
2211 self.adjust = adjust
2213 def i2m(self, pkt, x):
2214 # type: (Optional[Packet], Optional[int]) -> int
2215 if x is None and pkt is not None:
2216 if self.length_of is not None:
2217 fld, fval = pkt.getfield_and_val(self.length_of)
2218 f = fld.i2len(pkt, fval)
2219 elif self.count_of is not None:
2220 fld, fval = pkt.getfield_and_val(self.count_of)
2221 f = fld.i2count(pkt, fval)
2222 else:
2223 raise ValueError(
2224 "Field should have either length_of or count_of"
2225 )
2226 x = self.adjust(pkt, f)
2227 elif x is None:
2228 x = 0
2229 return x
2232class StrNullField(StrField):
2233 DELIMITER = b"\x00"
2235 def addfield(self, pkt, s, val):
2236 # type: (Packet, bytes, Optional[bytes]) -> bytes
2237 return s + self.i2m(pkt, val) + self.DELIMITER
2239 def getfield(self,
2240 pkt, # type: Packet
2241 s, # type: bytes
2242 ):
2243 # type: (...) -> Tuple[bytes, bytes]
2244 len_str = 0
2245 while True:
2246 len_str = s.find(self.DELIMITER, len_str)
2247 if len_str < 0:
2248 # DELIMITER not found: return empty
2249 return b"", s
2250 if len_str % len(self.DELIMITER):
2251 len_str += 1
2252 else:
2253 break
2254 return s[len_str + len(self.DELIMITER):], self.m2i(pkt, s[:len_str])
2256 def randval(self):
2257 # type: () -> RandTermString
2258 return RandTermString(RandNum(0, 1200), self.DELIMITER)
2260 def i2len(self, pkt, x):
2261 # type: (Optional[Packet], Any) -> int
2262 return super(StrNullField, self).i2len(pkt, x) + 1
2265class StrNullFieldUtf16(StrNullField, StrFieldUtf16):
2266 DELIMITER = b"\x00\x00"
2269class StrStopField(StrField):
2270 __slots__ = ["stop", "additional"]
2272 def __init__(self, name, default, stop, additional=0):
2273 # type: (str, str, bytes, int) -> None
2274 Field.__init__(self, name, default)
2275 self.stop = stop
2276 self.additional = additional
2278 def getfield(self, pkt, s):
2279 # type: (Optional[Packet], bytes) -> Tuple[bytes, bytes]
2280 len_str = s.find(self.stop)
2281 if len_str < 0:
2282 return b"", s
2283 len_str += len(self.stop) + self.additional
2284 return s[len_str:], s[:len_str]
2286 def randval(self):
2287 # type: () -> RandTermString
2288 return RandTermString(RandNum(0, 1200), self.stop)
2291class LenField(Field[int, int]):
2292 """
2293 If None, will be filled with the size of the payload
2294 """
2295 __slots__ = ["adjust"]
2297 def __init__(self, name, default, fmt="H", adjust=lambda x: x):
2298 # type: (str, Optional[Any], str, Callable[[int], int]) -> None
2299 Field.__init__(self, name, default, fmt)
2300 self.adjust = adjust
2302 def i2m(self,
2303 pkt, # type: Optional[Packet]
2304 x, # type: Optional[int]
2305 ):
2306 # type: (...) -> int
2307 if x is None:
2308 x = 0
2309 if pkt is not None:
2310 x = self.adjust(len(pkt.payload))
2311 return x
2314class BCDFloatField(Field[float, int]):
2315 def i2m(self, pkt, x):
2316 # type: (Optional[Packet], Optional[float]) -> int
2317 if x is None:
2318 return 0
2319 return int(256 * x)
2321 def m2i(self, pkt, x):
2322 # type: (Optional[Packet], int) -> float
2323 return x / 256.0
2326class _BitField(Field[I, int]):
2327 """
2328 Field to handle bits.
2330 :param name: name of the field
2331 :param default: default value
2332 :param size: size (in bits). If negative, Low endian
2333 :param tot_size: size of the total group of bits (in bytes) the bitfield
2334 is in. If negative, Low endian.
2335 :param end_tot_size: same but for the BitField ending a group.
2337 Example - normal usage::
2339 0 1 2 3
2340 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
2341 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
2342 | A | B | C |
2343 +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
2345 Fig. TestPacket
2347 class TestPacket(Packet):
2348 fields_desc = [
2349 BitField("a", 0, 14),
2350 BitField("b", 0, 16),
2351 BitField("c", 0, 2),
2352 ]
2354 Example - Low endian stored as 16 bits on the network::
2356 x x x x x x x x x x x x x x x x
2357 a [b] [ c ] [ a ]
2359 Will first get reversed during dissecion:
2361 x x x x x x x x x x x x x x x x
2362 [ a ] [b] [ c ]
2364 class TestPacket(Packet):
2365 fields_desc = [
2366 BitField("a", 0, 9, tot_size=-2),
2367 BitField("b", 0, 2),
2368 BitField("c", 0, 5, end_tot_size=-2)
2369 ]
2371 """
2372 __slots__ = ["rev", "size", "tot_size", "end_tot_size"]
2374 def __init__(self, name, default, size,
2375 tot_size=0, end_tot_size=0):
2376 # type: (str, Optional[I], int, int, int) -> None
2377 Field.__init__(self, name, default)
2378 if callable(size):
2379 size = size(self)
2380 self.rev = size < 0 or tot_size < 0 or end_tot_size < 0
2381 self.size = abs(size)
2382 if not tot_size:
2383 tot_size = self.size // 8
2384 self.tot_size = abs(tot_size)
2385 if not end_tot_size:
2386 end_tot_size = self.size // 8
2387 self.end_tot_size = abs(end_tot_size)
2388 # Fields always have a round sz except BitField
2389 # so to keep it simple, we'll ignore it here.
2390 self.sz = self.size / 8. # type: ignore
2392 # We need to # type: ignore a few things because of how special
2393 # BitField is
2394 def addfield(self, # type: ignore
2395 pkt, # type: Packet
2396 s, # type: Union[Tuple[bytes, int, int], bytes]
2397 ival, # type: I
2398 ):
2399 # type: (...) -> Union[Tuple[bytes, int, int], bytes]
2400 val = self.i2m(pkt, ival)
2401 if isinstance(s, tuple):
2402 s, bitsdone, v = s
2403 else:
2404 bitsdone = 0
2405 v = 0
2406 v <<= self.size
2407 v |= val & ((1 << self.size) - 1)
2408 bitsdone += self.size
2409 while bitsdone >= 8:
2410 bitsdone -= 8
2411 s = s + struct.pack("!B", v >> bitsdone)
2412 v &= (1 << bitsdone) - 1
2413 if bitsdone:
2414 return s, bitsdone, v
2415 else:
2416 # Apply LE if necessary
2417 if self.rev and self.end_tot_size > 1:
2418 s = s[:-self.end_tot_size] + s[-self.end_tot_size:][::-1]
2419 return s
2421 def getfield(self, # type: ignore
2422 pkt, # type: Packet
2423 s, # type: Union[Tuple[bytes, int], bytes]
2424 ):
2425 # type: (...) -> Union[Tuple[Tuple[bytes, int], I], Tuple[bytes, I]] # noqa: E501
2426 if isinstance(s, tuple):
2427 s, bn = s
2428 else:
2429 bn = 0
2430 # Apply LE if necessary
2431 if self.rev and self.tot_size > 1:
2432 s = s[:self.tot_size][::-1] + s[self.tot_size:]
2434 # we don't want to process all the string
2435 nb_bytes = (self.size + bn - 1) // 8 + 1
2436 w = s[:nb_bytes]
2438 if len(w) < nb_bytes:
2439 raise struct.error("unpack requires a buffer of %d bytes" % nb_bytes)
2440 b = int.from_bytes(w, byteorder="big")
2442 # get rid of high order bits
2443 b &= (1 << (nb_bytes * 8 - bn)) - 1
2445 # remove low order bits
2446 b = b >> (nb_bytes * 8 - self.size - bn)
2448 bn += self.size
2449 s = s[bn // 8:]
2450 bn = bn % 8
2451 b2 = self.m2i(pkt, b)
2452 if bn:
2453 return (s, bn), b2
2454 else:
2455 return s, b2
2457 def randval(self):
2458 # type: () -> RandNum
2459 return RandNum(0, 2**self.size - 1)
2461 def i2len(self, pkt, x): # type: ignore
2462 # type: (Optional[Packet], Optional[float]) -> float
2463 return float(self.size) / 8
2466class BitField(_BitField[int]):
2467 __doc__ = _BitField.__doc__
2470class BitLenField(BitField):
2471 __slots__ = ["length_from"]
2473 def __init__(self,
2474 name, # type: str
2475 default, # type: Optional[int]
2476 length_from # type: Callable[[Packet], int]
2477 ):
2478 # type: (...) -> None
2479 self.length_from = length_from
2480 super(BitLenField, self).__init__(name, default, 0)
2482 def getfield(self, # type: ignore
2483 pkt, # type: Packet
2484 s, # type: Union[Tuple[bytes, int], bytes]
2485 ):
2486 # type: (...) -> Union[Tuple[Tuple[bytes, int], int], Tuple[bytes, int]] # noqa: E501
2487 self.size = self.length_from(pkt)
2488 return super(BitLenField, self).getfield(pkt, s)
2490 def addfield(self, # type: ignore
2491 pkt, # type: Packet
2492 s, # type: Union[Tuple[bytes, int, int], bytes]
2493 val # type: int
2494 ):
2495 # type: (...) -> Union[Tuple[bytes, int, int], bytes]
2496 self.size = self.length_from(pkt)
2497 return super(BitLenField, self).addfield(pkt, s, val)
2500class BitFieldLenField(BitField):
2501 __slots__ = ["length_of", "count_of", "adjust", "tot_size", "end_tot_size"]
2503 def __init__(self,
2504 name, # type: str
2505 default, # type: Optional[int]
2506 size, # type: int
2507 length_of=None, # type: Optional[Union[Callable[[Optional[Packet]], int], str]] # noqa: E501
2508 count_of=None, # type: Optional[str]
2509 adjust=lambda pkt, x: x, # type: Callable[[Optional[Packet], int], int] # noqa: E501
2510 tot_size=0, # type: int
2511 end_tot_size=0, # type: int
2512 ):
2513 # type: (...) -> None
2514 super(BitFieldLenField, self).__init__(name, default, size,
2515 tot_size, end_tot_size)
2516 self.length_of = length_of
2517 self.count_of = count_of
2518 self.adjust = adjust
2520 def i2m(self, pkt, x):
2521 # type: (Optional[Packet], Optional[Any]) -> int
2522 return FieldLenField.i2m(self, pkt, x) # type: ignore
2525class XBitField(BitField):
2526 def i2repr(self, pkt, x):
2527 # type: (Optional[Packet], int) -> str
2528 return lhex(self.i2h(pkt, x))
2531_EnumType = Union[Dict[I, str], Dict[str, I], List[str], DADict[I, str], Type[Enum], Tuple[Callable[[I], str], Callable[[str], I]]] # noqa: E501
2534class _EnumField(Field[Union[List[I], I], I]):
2535 def __init__(self,
2536 name, # type: str
2537 default, # type: Optional[I]
2538 enum, # type: _EnumType[I]
2539 fmt="H", # type: str
2540 ):
2541 # type: (...) -> None
2542 """ Initializes enum fields.
2544 @param name: name of this field
2545 @param default: default value of this field
2546 @param enum: either an enum, a dict or a tuple of two callables.
2547 Dict keys are the internal values, while the dict
2548 values are the user-friendly representations. If the
2549 tuple is provided, the first callable receives the
2550 internal value as parameter and returns the
2551 user-friendly representation and the second callable
2552 does the converse. The first callable may return None
2553 to default to a literal string (repr()) representation.
2554 @param fmt: struct.pack format used to parse and serialize the
2555 internal value from and to machine representation.
2556 """
2557 if isinstance(enum, ObservableDict):
2558 cast(ObservableDict, enum).observe(self)
2560 if isinstance(enum, tuple):
2561 self.i2s_cb = enum[0] # type: Optional[Callable[[I], str]]
2562 self.s2i_cb = enum[1] # type: Optional[Callable[[str], I]]
2563 self.i2s = None # type: Optional[Dict[I, str]]
2564 self.s2i = None # type: Optional[Dict[str, I]]
2565 elif isinstance(enum, type) and issubclass(enum, Enum):
2566 # Python's Enum
2567 i2s = self.i2s = {}
2568 s2i = self.s2i = {}
2569 self.i2s_cb = None
2570 self.s2i_cb = None
2571 names = [x.name for x in enum]
2572 for n in names:
2573 value = enum[n].value
2574 i2s[value] = n
2575 s2i[n] = value
2576 else:
2577 i2s = self.i2s = {}
2578 s2i = self.s2i = {}
2579 self.i2s_cb = None
2580 self.s2i_cb = None
2581 keys = [] # type: List[I]
2582 if isinstance(enum, list):
2583 keys = list(range(len(enum))) # type: ignore
2584 elif isinstance(enum, DADict):
2585 keys = enum.keys()
2586 else:
2587 keys = list(enum) # type: ignore
2588 if any(isinstance(x, str) for x in keys):
2589 i2s, s2i = s2i, i2s # type: ignore
2590 for k in keys:
2591 value = cast(str, enum[k]) # type: ignore
2592 i2s[k] = value
2593 s2i[value] = k
2594 Field.__init__(self, name, default, fmt)
2596 def any2i_one(self, pkt, x):
2597 # type: (Optional[Packet], Any) -> I
2598 if isinstance(x, Enum):
2599 return cast(I, x.value)
2600 elif isinstance(x, str):
2601 if self.s2i:
2602 x = self.s2i[x]
2603 elif self.s2i_cb:
2604 x = self.s2i_cb(x)
2605 return cast(I, x)
2607 def _i2repr(self, pkt, x):
2608 # type: (Optional[Packet], I) -> str
2609 return repr(x)
2611 def i2repr_one(self, pkt, x):
2612 # type: (Optional[Packet], I) -> str
2613 if self not in conf.noenum and not isinstance(x, VolatileValue):
2614 if self.i2s:
2615 try:
2616 return self.i2s[x]
2617 except KeyError:
2618 pass
2619 elif self.i2s_cb:
2620 ret = self.i2s_cb(x)
2621 if ret is not None:
2622 return ret
2623 return self._i2repr(pkt, x)
2625 def any2i(self, pkt, x):
2626 # type: (Optional[Packet], Any) -> Union[I, List[I]]
2627 if isinstance(x, list):
2628 return [self.any2i_one(pkt, z) for z in x]
2629 else:
2630 return self.any2i_one(pkt, x)
2632 def i2repr(self, pkt, x): # type: ignore
2633 # type: (Optional[Packet], Any) -> Union[List[str], str]
2634 if isinstance(x, list):
2635 return [self.i2repr_one(pkt, z) for z in x]
2636 else:
2637 return self.i2repr_one(pkt, x)
2639 def notify_set(self, enum, key, value):
2640 # type: (ObservableDict, I, str) -> None
2641 ks = "0x%x" if isinstance(key, int) else "%s"
2642 log_runtime.debug(
2643 "At %s: Change to %s at " + ks, self, value, key
2644 )
2645 if self.i2s is not None and self.s2i is not None:
2646 self.i2s[key] = value
2647 self.s2i[value] = key
2649 def notify_del(self, enum, key):
2650 # type: (ObservableDict, I) -> None
2651 ks = "0x%x" if isinstance(key, int) else "%s"
2652 log_runtime.debug("At %s: Delete value at " + ks, self, key)
2653 if self.i2s is not None and self.s2i is not None:
2654 value = self.i2s[key]
2655 del self.i2s[key]
2656 del self.s2i[value]
2659class EnumField(_EnumField[I]):
2660 __slots__ = ["i2s", "s2i", "s2i_cb", "i2s_cb"]
2663class CharEnumField(EnumField[str]):
2664 def __init__(self,
2665 name, # type: str
2666 default, # type: str
2667 enum, # type: _EnumType[str]
2668 fmt="1s", # type: str
2669 ):
2670 # type: (...) -> None
2671 super(CharEnumField, self).__init__(name, default, enum, fmt)
2672 if self.i2s is not None:
2673 k = list(self.i2s)
2674 if k and len(k[0]) != 1:
2675 self.i2s, self.s2i = self.s2i, self.i2s
2677 def any2i_one(self, pkt, x):
2678 # type: (Optional[Packet], str) -> str
2679 if len(x) != 1:
2680 if self.s2i:
2681 x = self.s2i[x]
2682 elif self.s2i_cb:
2683 x = self.s2i_cb(x)
2684 return x
2687class BitEnumField(_BitField[Union[List[int], int]], _EnumField[int]):
2688 __slots__ = EnumField.__slots__
2690 def __init__(self,
2691 name, # type: str
2692 default, # type: Optional[int]
2693 size, # type: int
2694 enum, # type: _EnumType[int]
2695 **kwargs # type: Any
2696 ):
2697 # type: (...) -> None
2698 _EnumField.__init__(self, name, default, enum)
2699 _BitField.__init__(self, name, default, size, **kwargs)
2701 def any2i(self, pkt, x):
2702 # type: (Optional[Packet], Any) -> Union[List[int], int]
2703 return _EnumField.any2i(self, pkt, x)
2705 def i2repr(self,
2706 pkt, # type: Optional[Packet]
2707 x, # type: Union[List[int], int]
2708 ):
2709 # type: (...) -> Any
2710 return _EnumField.i2repr(self, pkt, x)
2713class BitLenEnumField(BitLenField, _EnumField[int]):
2714 __slots__ = EnumField.__slots__
2716 def __init__(self,
2717 name, # type: str
2718 default, # type: Optional[int]
2719 length_from, # type: Callable[[Packet], int]
2720 enum, # type: _EnumType[int]
2721 **kwargs, # type: Any
2722 ):
2723 # type: (...) -> None
2724 _EnumField.__init__(self, name, default, enum)
2725 BitLenField.__init__(self, name, default, length_from, **kwargs)
2727 def any2i(self, pkt, x):
2728 # type: (Optional[Packet], Any) -> int
2729 return _EnumField.any2i(self, pkt, x) # type: ignore
2731 def i2repr(self,
2732 pkt, # type: Optional[Packet]
2733 x, # type: Union[List[int], int]
2734 ):
2735 # type: (...) -> Any
2736 return _EnumField.i2repr(self, pkt, x)
2739class ShortEnumField(EnumField[int]):
2740 __slots__ = EnumField.__slots__
2742 def __init__(self,
2743 name, # type: str
2744 default, # type: Optional[int]
2745 enum, # type: _EnumType[int]
2746 ):
2747 # type: (...) -> None
2748 super(ShortEnumField, self).__init__(name, default, enum, "H")
2751class LEShortEnumField(EnumField[int]):
2752 def __init__(self,
2753 name, # type: str
2754 default, # type: Optional[int]
2755 enum, # type: _EnumType[int]
2756 ):
2757 # type: (...) -> None
2758 super(LEShortEnumField, self).__init__(name, default, enum, "<H")
2761class LongEnumField(EnumField[int]):
2762 def __init__(self,
2763 name, # type: str
2764 default, # type: Optional[int]
2765 enum, # type: _EnumType[int]
2766 ):
2767 # type: (...) -> None
2768 super(LongEnumField, self).__init__(name, default, enum, "Q")
2771class LELongEnumField(EnumField[int]):
2772 def __init__(self,
2773 name, # type: str
2774 default, # type: Optional[int]
2775 enum, # type: _EnumType[int]
2776 ):
2777 # type: (...) -> None
2778 super(LELongEnumField, self).__init__(name, default, enum, "<Q")
2781class ByteEnumField(EnumField[int]):
2782 def __init__(self,
2783 name, # type: str
2784 default, # type: Optional[int]
2785 enum, # type: _EnumType[int]
2786 ):
2787 # type: (...) -> None
2788 super(ByteEnumField, self).__init__(name, default, enum, "B")
2791class XByteEnumField(ByteEnumField):
2792 def i2repr_one(self, pkt, x):
2793 # type: (Optional[Packet], int) -> str
2794 if self not in conf.noenum and not isinstance(x, VolatileValue):
2795 if self.i2s:
2796 try:
2797 return self.i2s[x]
2798 except KeyError:
2799 pass
2800 elif self.i2s_cb:
2801 ret = self.i2s_cb(x)
2802 if ret is not None:
2803 return ret
2804 return lhex(x)
2807class IntEnumField(EnumField[int]):
2808 def __init__(self,
2809 name, # type: str
2810 default, # type: Optional[int]
2811 enum, # type: _EnumType[int]
2812 ):
2813 # type: (...) -> None
2814 super(IntEnumField, self).__init__(name, default, enum, "I")
2817class SignedIntEnumField(EnumField[int]):
2818 def __init__(self,
2819 name, # type: str
2820 default, # type: Optional[int]
2821 enum, # type: _EnumType[int]
2822 ):
2823 # type: (...) -> None
2824 super(SignedIntEnumField, self).__init__(name, default, enum, "i")
2827class LEIntEnumField(EnumField[int]):
2828 def __init__(self,
2829 name, # type: str
2830 default, # type: Optional[int]
2831 enum, # type: _EnumType[int]
2832 ):
2833 # type: (...) -> None
2834 super(LEIntEnumField, self).__init__(name, default, enum, "<I")
2837class XLEIntEnumField(LEIntEnumField):
2838 def _i2repr(self, pkt, x):
2839 # type: (Optional[Packet], Any) -> str
2840 return lhex(x)
2843class XShortEnumField(ShortEnumField):
2844 def _i2repr(self, pkt, x):
2845 # type: (Optional[Packet], Any) -> str
2846 return lhex(x)
2849class LE3BytesEnumField(LEThreeBytesField, _EnumField[int]):
2850 __slots__ = EnumField.__slots__
2852 def __init__(self,
2853 name, # type: str
2854 default, # type: Optional[int]
2855 enum, # type: _EnumType[int]
2856 ):
2857 # type: (...) -> None
2858 _EnumField.__init__(self, name, default, enum)
2859 LEThreeBytesField.__init__(self, name, default)
2861 def any2i(self, pkt, x):
2862 # type: (Optional[Packet], Any) -> int
2863 return _EnumField.any2i(self, pkt, x) # type: ignore
2865 def i2repr(self, pkt, x): # type: ignore
2866 # type: (Optional[Packet], Any) -> Union[List[str], str]
2867 return _EnumField.i2repr(self, pkt, x)
2870class XLE3BytesEnumField(LE3BytesEnumField):
2871 def _i2repr(self, pkt, x):
2872 # type: (Optional[Packet], Any) -> str
2873 return lhex(x)
2876class _MultiEnumField(_EnumField[I]):
2877 def __init__(self,
2878 name, # type: str
2879 default, # type: int
2880 enum, # type: Dict[I, Dict[I, str]]
2881 depends_on, # type: Callable[[Optional[Packet]], I]
2882 fmt="H" # type: str
2883 ):
2884 # type: (...) -> None
2886 self.depends_on = depends_on
2887 self.i2s_multi = enum
2888 self.s2i_multi = {} # type: Dict[I, Dict[str, I]]
2889 self.s2i_all = {} # type: Dict[str, I]
2890 for m in enum:
2891 s2i = {} # type: Dict[str, I]
2892 self.s2i_multi[m] = s2i
2893 for k, v in enum[m].items():
2894 s2i[v] = k
2895 self.s2i_all[v] = k
2896 Field.__init__(self, name, default, fmt)
2898 def any2i_one(self, pkt, x):
2899 # type: (Optional[Packet], Any) -> I
2900 if isinstance(x, str):
2901 v = self.depends_on(pkt)
2902 if v in self.s2i_multi:
2903 s2i = self.s2i_multi[v]
2904 if x in s2i:
2905 return s2i[x]
2906 return self.s2i_all[x]
2907 return cast(I, x)
2909 def i2repr_one(self, pkt, x):
2910 # type: (Optional[Packet], I) -> str
2911 v = self.depends_on(pkt)
2912 if isinstance(v, VolatileValue):
2913 return repr(v)
2914 if v in self.i2s_multi:
2915 return str(self.i2s_multi[v].get(x, x))
2916 return str(x)
2919class MultiEnumField(_MultiEnumField[int], EnumField[int]):
2920 __slots__ = ["depends_on", "i2s_multi", "s2i_multi", "s2i_all"]
2923class BitMultiEnumField(_BitField[Union[List[int], int]],
2924 _MultiEnumField[int]):
2925 __slots__ = EnumField.__slots__ + MultiEnumField.__slots__
2927 def __init__(
2928 self,
2929 name, # type: str
2930 default, # type: int
2931 size, # type: int
2932 enum, # type: Dict[int, Dict[int, str]]
2933 depends_on # type: Callable[[Optional[Packet]], int]
2934 ):
2935 # type: (...) -> None
2936 _MultiEnumField.__init__(self, name, default, enum, depends_on)
2937 self.rev = size < 0
2938 self.size = abs(size)
2939 self.sz = self.size / 8. # type: ignore
2941 def any2i(self, pkt, x):
2942 # type: (Optional[Packet], Any) -> Union[List[int], int]
2943 return _MultiEnumField[int].any2i(
2944 self, # type: ignore
2945 pkt,
2946 x
2947 )
2949 def i2repr( # type: ignore
2950 self,
2951 pkt, # type: Optional[Packet]
2952 x # type: Union[List[int], int]
2953 ):
2954 # type: (...) -> Union[str, List[str]]
2955 return _MultiEnumField[int].i2repr(
2956 self, # type: ignore
2957 pkt,
2958 x
2959 )
2962class ByteEnumKeysField(ByteEnumField):
2963 """ByteEnumField that picks valid values when fuzzed. """
2965 def randval(self):
2966 # type: () -> RandEnumKeys
2967 return RandEnumKeys(self.i2s or {})
2970class ShortEnumKeysField(ShortEnumField):
2971 """ShortEnumField that picks valid values when fuzzed. """
2973 def randval(self):
2974 # type: () -> RandEnumKeys
2975 return RandEnumKeys(self.i2s or {})
2978class IntEnumKeysField(IntEnumField):
2979 """IntEnumField that picks valid values when fuzzed. """
2981 def randval(self):
2982 # type: () -> RandEnumKeys
2983 return RandEnumKeys(self.i2s or {})
2986# Little endian fixed length field
2989class LEFieldLenField(FieldLenField):
2990 def __init__(
2991 self,
2992 name, # type: str
2993 default, # type: Optional[Any]
2994 length_of=None, # type: Optional[str]
2995 fmt="<H", # type: str
2996 count_of=None, # type: Optional[str]
2997 adjust=lambda pkt, x: x, # type: Callable[[Packet, int], int]
2998 ):
2999 # type: (...) -> None
3000 FieldLenField.__init__(
3001 self, name, default,
3002 length_of=length_of,
3003 fmt=fmt,
3004 count_of=count_of,
3005 adjust=adjust
3006 )
3009class FlagValueIter(object):
3011 __slots__ = ["flagvalue", "cursor"]
3013 def __init__(self, flagvalue):
3014 # type: (FlagValue) -> None
3015 self.flagvalue = flagvalue
3016 self.cursor = 0
3018 def __iter__(self):
3019 # type: () -> FlagValueIter
3020 return self
3022 def __next__(self):
3023 # type: () -> str
3024 x = int(self.flagvalue)
3025 x >>= self.cursor
3026 while x:
3027 self.cursor += 1
3028 if x & 1:
3029 return self.flagvalue.names[self.cursor - 1]
3030 x >>= 1
3031 raise StopIteration
3033 next = __next__
3036class FlagValue(object):
3037 __slots__ = ["value", "names", "multi"]
3039 def _fixvalue(self, value):
3040 # type: (Any) -> int
3041 if not value:
3042 return 0
3043 if isinstance(value, str):
3044 value = value.split('+') if self.multi else list(value)
3045 if isinstance(value, list):
3046 y = 0
3047 for i in value:
3048 y |= 1 << self.names.index(i)
3049 value = y
3050 return int(value)
3052 def __init__(self, value, names):
3053 # type: (Union[List[str], int, str], Union[List[str], str]) -> None
3054 self.multi = isinstance(names, list)
3055 self.names = names
3056 self.value = self._fixvalue(value)
3058 def __hash__(self):
3059 # type: () -> int
3060 return hash(self.value)
3062 def __int__(self):
3063 # type: () -> int
3064 return self.value
3066 def __eq__(self, other):
3067 # type: (Any) -> bool
3068 return self.value == self._fixvalue(other)
3070 def __lt__(self, other):
3071 # type: (Any) -> bool
3072 return self.value < self._fixvalue(other)
3074 def __le__(self, other):
3075 # type: (Any) -> bool
3076 return self.value <= self._fixvalue(other)
3078 def __gt__(self, other):
3079 # type: (Any) -> bool
3080 return self.value > self._fixvalue(other)
3082 def __ge__(self, other):
3083 # type: (Any) -> bool
3084 return self.value >= self._fixvalue(other)
3086 def __ne__(self, other):
3087 # type: (Any) -> bool
3088 return self.value != self._fixvalue(other)
3090 def __and__(self, other):
3091 # type: (int) -> FlagValue
3092 return self.__class__(self.value & self._fixvalue(other), self.names)
3093 __rand__ = __and__
3095 def __or__(self, other):
3096 # type: (int) -> FlagValue
3097 return self.__class__(self.value | self._fixvalue(other), self.names)
3098 __ror__ = __or__
3099 __add__ = __or__ # + is an alias for |
3101 def __sub__(self, other):
3102 # type: (int) -> FlagValue
3103 return self.__class__(
3104 self.value & (2 ** len(self.names) - 1 - self._fixvalue(other)),
3105 self.names
3106 )
3108 def __xor__(self, other):
3109 # type: (int) -> FlagValue
3110 return self.__class__(self.value ^ self._fixvalue(other), self.names)
3112 def __lshift__(self, other):
3113 # type: (int) -> int
3114 return self.value << self._fixvalue(other)
3116 def __rshift__(self, other):
3117 # type: (int) -> int
3118 return self.value >> self._fixvalue(other)
3120 def __nonzero__(self):
3121 # type: () -> bool
3122 return bool(self.value)
3123 __bool__ = __nonzero__
3125 def flagrepr(self):
3126 # type: () -> str
3127 warnings.warn(
3128 "obj.flagrepr() is obsolete. Use str(obj) instead.",
3129 DeprecationWarning
3130 )
3131 return str(self)
3133 def __str__(self):
3134 # type: () -> str
3135 i = 0
3136 r = []
3137 x = int(self)
3138 while x:
3139 if x & 1:
3140 try:
3141 name = self.names[i]
3142 except IndexError:
3143 name = "?"
3144 r.append(name)
3145 i += 1
3146 x >>= 1
3147 return ("+" if self.multi else "").join(r)
3149 def __iter__(self):
3150 # type: () -> FlagValueIter
3151 return FlagValueIter(self)
3153 def __repr__(self):
3154 # type: () -> str
3155 return "<Flag %d (%s)>" % (self, self)
3157 def __deepcopy__(self, memo):
3158 # type: (Dict[Any, Any]) -> FlagValue
3159 return self.__class__(int(self), self.names)
3161 def __getattr__(self, attr):
3162 # type: (str) -> Any
3163 if attr in self.__slots__:
3164 return super(FlagValue, self).__getattribute__(attr)
3165 try:
3166 if self.multi:
3167 return bool((2 ** self.names.index(attr)) & int(self))
3168 return all(bool((2 ** self.names.index(flag)) & int(self))
3169 for flag in attr)
3170 except ValueError:
3171 if '_' in attr:
3172 try:
3173 return self.__getattr__(attr.replace('_', '-'))
3174 except AttributeError:
3175 pass
3176 return super(FlagValue, self).__getattribute__(attr)
3178 def __setattr__(self, attr, value):
3179 # type: (str, Union[List[str], int, str]) -> None
3180 if attr == "value" and not isinstance(value, int):
3181 raise ValueError(value)
3182 if attr in self.__slots__:
3183 return super(FlagValue, self).__setattr__(attr, value)
3184 if attr in self.names:
3185 if value:
3186 self.value |= (2 ** self.names.index(attr))
3187 else:
3188 self.value &= ~(2 ** self.names.index(attr))
3189 else:
3190 return super(FlagValue, self).__setattr__(attr, value)
3192 def copy(self):
3193 # type: () -> FlagValue
3194 return self.__class__(self.value, self.names)
3197class FlagsField(_BitField[Optional[Union[int, FlagValue]]]):
3198 """ Handle Flag type field
3200 Make sure all your flags have a label
3202 Example (list):
3203 >>> from scapy.packet import Packet
3204 >>> class FlagsTest(Packet):
3205 fields_desc = [FlagsField("flags", 0, 8, ["f0", "f1", "f2", "f3", "f4", "f5", "f6", "f7"])] # noqa: E501
3206 >>> FlagsTest(flags=9).show2()
3207 ###[ FlagsTest ]###
3208 flags = f0+f3
3210 Example (str):
3211 >>> from scapy.packet import Packet
3212 >>> class TCPTest(Packet):
3213 fields_desc = [
3214 BitField("reserved", 0, 7),
3215 FlagsField("flags", 0x2, 9, "FSRPAUECN")
3216 ]
3217 >>> TCPTest(flags=3).show2()
3218 ###[ FlagsTest ]###
3219 reserved = 0
3220 flags = FS
3222 Example (dict):
3223 >>> from scapy.packet import Packet
3224 >>> class FlagsTest2(Packet):
3225 fields_desc = [
3226 FlagsField("flags", 0x2, 16, {
3227 0x0001: "A",
3228 0x0008: "B",
3229 })
3230 ]
3232 :param name: field's name
3233 :param default: default value for the field
3234 :param size: number of bits in the field (in bits). if negative, LE
3235 :param names: (list or str or dict) label for each flag
3236 If it's a str or a list, the least Significant Bit tag's name
3237 is written first.
3238 """
3239 ismutable = True
3240 __slots__ = ["names"]
3242 def __init__(self,
3243 name, # type: str
3244 default, # type: Optional[Union[int, FlagValue]]
3245 size, # type: int
3246 names, # type: Union[List[str], str, Dict[int, str]]
3247 **kwargs # type: Any
3248 ):
3249 # type: (...) -> None
3250 # Convert the dict to a list
3251 if isinstance(names, dict):
3252 tmp = ["bit_%d" % i for i in range(abs(size))]
3253 for i, v in names.items():
3254 tmp[int(math.floor(math.log(i, 2)))] = v
3255 names = tmp
3256 # Store the names as str or list
3257 self.names = names
3258 super(FlagsField, self).__init__(name, default, size, **kwargs)
3260 def _fixup_val(self, x):
3261 # type: (Any) -> Optional[FlagValue]
3262 """Returns a FlagValue instance when needed. Internal method, to be
3263used in *2i() and i2*() methods.
3265 """
3266 if isinstance(x, (FlagValue, VolatileValue)):
3267 return x # type: ignore
3268 if x is None:
3269 return None
3270 return FlagValue(x, self.names)
3272 def any2i(self, pkt, x):
3273 # type: (Optional[Packet], Any) -> Optional[FlagValue]
3274 return self._fixup_val(super(FlagsField, self).any2i(pkt, x))
3276 def m2i(self, pkt, x):
3277 # type: (Optional[Packet], int) -> Optional[FlagValue]
3278 return self._fixup_val(super(FlagsField, self).m2i(pkt, x))
3280 def i2h(self, pkt, x):
3281 # type: (Optional[Packet], Any) -> Optional[FlagValue]
3282 return self._fixup_val(super(FlagsField, self).i2h(pkt, x))
3284 def i2repr(self,
3285 pkt, # type: Optional[Packet]
3286 x, # type: Any
3287 ):
3288 # type: (...) -> str
3289 if isinstance(x, (list, tuple)):
3290 return repr(type(x)(
3291 "None" if v is None else str(self._fixup_val(v)) for v in x
3292 ))
3293 return "None" if x is None else str(self._fixup_val(x))
3296MultiFlagsEntry = collections.namedtuple('MultiFlagsEntry', ['short', 'long'])
3299class MultiFlagsField(_BitField[Set[str]]):
3300 __slots__ = FlagsField.__slots__ + ["depends_on"]
3302 def __init__(self,
3303 name, # type: str
3304 default, # type: Set[str]
3305 size, # type: int
3306 names, # type: Dict[int, Dict[int, MultiFlagsEntry]]
3307 depends_on, # type: Callable[[Optional[Packet]], int]
3308 ):
3309 # type: (...) -> None
3310 self.names = names
3311 self.depends_on = depends_on
3312 super(MultiFlagsField, self).__init__(name, default, size)
3314 def any2i(self, pkt, x):
3315 # type: (Optional[Packet], Any) -> Set[str]
3316 if not isinstance(x, (set, int)):
3317 raise ValueError('set expected')
3319 if pkt is not None:
3320 if isinstance(x, int):
3321 return self.m2i(pkt, x)
3322 else:
3323 v = self.depends_on(pkt)
3324 if v is not None:
3325 assert v in self.names, 'invalid dependency'
3326 these_names = self.names[v]
3327 s = set()
3328 for i in x:
3329 for val in these_names.values():
3330 if val.short == i:
3331 s.add(i)
3332 break
3333 else:
3334 assert False, 'Unknown flag "{}" with this dependency'.format(i) # noqa: E501
3335 continue
3336 return s
3337 if isinstance(x, int):
3338 return set()
3339 return x
3341 def i2m(self, pkt, x):
3342 # type: (Optional[Packet], Optional[Set[str]]) -> int
3343 v = self.depends_on(pkt)
3344 these_names = self.names.get(v, {})
3346 r = 0
3347 if x is None:
3348 return r
3349 for flag_set in x:
3350 for i, val in these_names.items():
3351 if val.short == flag_set:
3352 r |= 1 << i
3353 break
3354 else:
3355 r |= 1 << int(flag_set[len('bit '):])
3356 return r
3358 def m2i(self, pkt, x):
3359 # type: (Optional[Packet], int) -> Set[str]
3360 v = self.depends_on(pkt)
3361 these_names = self.names.get(v, {})
3363 r = set()
3364 i = 0
3365 while x:
3366 if x & 1:
3367 if i in these_names:
3368 r.add(these_names[i].short)
3369 else:
3370 r.add('bit {}'.format(i))
3371 x >>= 1
3372 i += 1
3373 return r
3375 def i2repr(self, pkt, x):
3376 # type: (Optional[Packet], Set[str]) -> str
3377 v = self.depends_on(pkt)
3378 these_names = self.names.get(v, {})
3380 r = set()
3381 for flag_set in x:
3382 for i in these_names.values():
3383 if i.short == flag_set:
3384 r.add("{} ({})".format(i.long, i.short))
3385 break
3386 else:
3387 r.add(flag_set)
3388 return repr(r)
3391class FixedPointField(BitField):
3392 __slots__ = ['frac_bits']
3394 def __init__(self, name, default, size, frac_bits=16):
3395 # type: (str, int, int, int) -> None
3396 self.frac_bits = frac_bits
3397 super(FixedPointField, self).__init__(name, default, size)
3399 def any2i(self, pkt, val):
3400 # type: (Optional[Packet], Optional[float]) -> Optional[int]
3401 if val is None:
3402 return val
3403 ival = int(val)
3404 fract = int((val - ival) * 2**self.frac_bits)
3405 return (ival << self.frac_bits) | fract
3407 def i2h(self, pkt, val):
3408 # type: (Optional[Packet], Optional[int]) -> Optional[EDecimal]
3409 # A bit of trickery to get precise floats
3410 if val is None:
3411 return val
3412 int_part = val >> self.frac_bits
3413 pw = 2.0**self.frac_bits
3414 frac_part = EDecimal(val & (1 << self.frac_bits) - 1)
3415 frac_part /= pw # type: ignore
3416 return int_part + frac_part.normalize(int(math.log10(pw)))
3418 def i2repr(self, pkt, val):
3419 # type: (Optional[Packet], int) -> str
3420 return str(self.i2h(pkt, val))
3423# Base class for IPv4 and IPv6 Prefixes inspired by IPField and IP6Field.
3424# Machine values are encoded in a multiple of wordbytes bytes.
3425class _IPPrefixFieldBase(Field[Tuple[str, int], Tuple[bytes, int]]):
3426 __slots__ = ["wordbytes", "maxbytes", "aton", "ntoa", "length_from"]
3428 def __init__(
3429 self,
3430 name, # type: str
3431 default, # type: Tuple[str, int]
3432 wordbytes, # type: int
3433 maxbytes, # type: int
3434 aton, # type: Callable[..., Any]
3435 ntoa, # type: Callable[..., Any]
3436 length_from=None # type: Optional[Callable[[Packet], int]]
3437 ):
3438 # type: (...) -> None
3439 self.wordbytes = wordbytes
3440 self.maxbytes = maxbytes
3441 self.aton = aton
3442 self.ntoa = ntoa
3443 Field.__init__(self, name, default, "%is" % self.maxbytes)
3444 if length_from is None:
3445 length_from = lambda x: 0
3446 self.length_from = length_from
3448 def _numbytes(self, pfxlen):
3449 # type: (int) -> int
3450 wbits = self.wordbytes * 8
3451 return ((pfxlen + (wbits - 1)) // wbits) * self.wordbytes
3453 def h2i(self, pkt, x):
3454 # type: (Optional[Packet], str) -> Tuple[str, int]
3455 # "fc00:1::1/64" -> ("fc00:1::1", 64)
3456 [pfx, pfxlen] = x.split('/')
3457 self.aton(pfx) # check for validity
3458 return (pfx, int(pfxlen))
3460 def i2h(self, pkt, x):
3461 # type: (Optional[Packet], Tuple[str, int]) -> str
3462 # ("fc00:1::1", 64) -> "fc00:1::1/64"
3463 (pfx, pfxlen) = x
3464 return "%s/%i" % (pfx, pfxlen)
3466 def i2m(self,
3467 pkt, # type: Optional[Packet]
3468 x # type: Optional[Tuple[str, int]]
3469 ):
3470 # type: (...) -> Tuple[bytes, int]
3471 # ("fc00:1::1", 64) -> (b"\xfc\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01", 64) # noqa: E501
3472 if x is None:
3473 pfx, pfxlen = "", 0
3474 else:
3475 (pfx, pfxlen) = x
3476 s = self.aton(pfx)
3477 return (s[:self._numbytes(pfxlen)], pfxlen)
3479 def m2i(self, pkt, x):
3480 # type: (Optional[Packet], Tuple[bytes, int]) -> Tuple[str, int]
3481 # (b"\xfc\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01", 64) -> ("fc00:1::1", 64) # noqa: E501
3482 (s, pfxlen) = x
3484 if len(s) < self.maxbytes:
3485 s = s + (b"\0" * (self.maxbytes - len(s)))
3486 return (self.ntoa(s), pfxlen)
3488 def any2i(self, pkt, x):
3489 # type: (Optional[Packet], Optional[Any]) -> Tuple[str, int]
3490 if x is None:
3491 return (self.ntoa(b"\0" * self.maxbytes), 1)
3493 return self.h2i(pkt, x)
3495 def i2len(self, pkt, x):
3496 # type: (Packet, Tuple[str, int]) -> int
3497 (_, pfxlen) = x
3498 return pfxlen
3500 def addfield(self, pkt, s, val):
3501 # type: (Packet, bytes, Optional[Tuple[str, int]]) -> bytes
3502 (rawpfx, pfxlen) = self.i2m(pkt, val)
3503 fmt = "!%is" % self._numbytes(pfxlen)
3504 return s + struct.pack(fmt, rawpfx)
3506 def getfield(self, pkt, s):
3507 # type: (Packet, bytes) -> Tuple[bytes, Tuple[str, int]]
3508 pfxlen = self.length_from(pkt)
3509 numbytes = self._numbytes(pfxlen)
3510 fmt = "!%is" % numbytes
3511 return s[numbytes:], self.m2i(pkt, (struct.unpack(fmt, s[:numbytes])[0], pfxlen)) # noqa: E501
3514class IPPrefixField(_IPPrefixFieldBase):
3515 def __init__(
3516 self,
3517 name, # type: str
3518 default, # type: Tuple[str, int]
3519 wordbytes=1, # type: int
3520 length_from=None # type: Optional[Callable[[Packet], int]]
3521 ):
3522 _IPPrefixFieldBase.__init__(
3523 self,
3524 name,
3525 default,
3526 wordbytes,
3527 4,
3528 inet_aton,
3529 inet_ntoa,
3530 length_from
3531 )
3534class IP6PrefixField(_IPPrefixFieldBase):
3535 def __init__(
3536 self,
3537 name, # type: str
3538 default, # type: Tuple[str, int]
3539 wordbytes=1, # type: int
3540 length_from=None # type: Optional[Callable[[Packet], int]]
3541 ):
3542 # type: (...) -> None
3543 _IPPrefixFieldBase.__init__(
3544 self,
3545 name,
3546 default,
3547 wordbytes,
3548 16,
3549 lambda a: inet_pton(socket.AF_INET6, a),
3550 lambda n: inet_ntop(socket.AF_INET6, n),
3551 length_from
3552 )
3555class UTCTimeField(Field[float, int]):
3556 __slots__ = ["epoch", "delta", "strf",
3557 "use_msec", "use_micro", "use_nano", "custom_scaling"]
3559 def __init__(self,
3560 name, # type: str
3561 default, # type: int
3562 use_msec=False, # type: bool
3563 use_micro=False, # type: bool
3564 use_nano=False, # type: bool
3565 epoch=None, # type: Optional[Tuple[int, int, int, int, int, int, int, int, int]] # noqa: E501
3566 strf="%a, %d %b %Y %H:%M:%S %z", # type: str
3567 custom_scaling=None, # type: Optional[int]
3568 fmt="I" # type: str
3569 ):
3570 # type: (...) -> None
3571 Field.__init__(self, name, default, fmt=fmt)
3572 mk_epoch = EPOCH if epoch is None else calendar.timegm(epoch)
3573 self.epoch = mk_epoch
3574 self.delta = mk_epoch - EPOCH
3575 self.strf = strf
3576 self.use_msec = use_msec
3577 self.use_micro = use_micro
3578 self.use_nano = use_nano
3579 self.custom_scaling = custom_scaling
3581 def i2repr(self, pkt, x):
3582 # type: (Optional[Packet], float) -> str
3583 if x is None:
3584 x = time.time() - self.delta
3585 elif self.use_msec:
3586 x = x / 1e3
3587 elif self.use_micro:
3588 x = x / 1e6
3589 elif self.use_nano:
3590 x = x / 1e9
3591 elif self.custom_scaling:
3592 x = x / self.custom_scaling
3593 x += self.delta
3594 # To make negative timestamps work on all plateforms (e.g. Windows),
3595 # we need a trick.
3596 t = (
3597 datetime.datetime(1970, 1, 1) +
3598 datetime.timedelta(seconds=x)
3599 ).strftime(self.strf)
3600 return "%s (%d)" % (t, int(x))
3602 def i2m(self, pkt, x):
3603 # type: (Optional[Packet], Optional[float]) -> int
3604 if x is None:
3605 x = time.time() - self.delta
3606 if self.use_msec:
3607 x = x * 1e3
3608 elif self.use_micro:
3609 x = x * 1e6
3610 elif self.use_nano:
3611 x = x * 1e9
3612 elif self.custom_scaling:
3613 x = x * self.custom_scaling
3614 return int(x)
3615 return int(x)
3618class SecondsIntField(Field[float, int]):
3619 __slots__ = ["use_msec", "use_micro", "use_nano"]
3621 def __init__(self, name, default,
3622 use_msec=False,
3623 use_micro=False,
3624 use_nano=False):
3625 # type: (str, int, bool, bool, bool) -> None
3626 Field.__init__(self, name, default, "I")
3627 self.use_msec = use_msec
3628 self.use_micro = use_micro
3629 self.use_nano = use_nano
3631 def i2repr(self, pkt, x):
3632 # type: (Optional[Packet], Optional[float]) -> str
3633 if x is None:
3634 y = 0 # type: Union[int, float]
3635 elif self.use_msec:
3636 y = x / 1e3
3637 elif self.use_micro:
3638 y = x / 1e6
3639 elif self.use_nano:
3640 y = x / 1e9
3641 else:
3642 y = x
3643 return "%s sec" % y
3646class _ScalingField(object):
3647 def __init__(self,
3648 name, # type: str
3649 default, # type: float
3650 scaling=1, # type: Union[int, float]
3651 unit="", # type: str
3652 offset=0, # type: Union[int, float]
3653 ndigits=3, # type: int
3654 fmt="B", # type: str
3655 ):
3656 # type: (...) -> None
3657 self.scaling = scaling
3658 self.unit = unit
3659 self.offset = offset
3660 self.ndigits = ndigits
3661 Field.__init__(self, name, default, fmt) # type: ignore
3663 def i2m(self,
3664 pkt, # type: Optional[Packet]
3665 x # type: Optional[Union[int, float]]
3666 ):
3667 # type: (...) -> Union[int, float]
3668 if x is None:
3669 x = 0
3670 x = (x - self.offset) / self.scaling
3671 if isinstance(x, float) and self.fmt[-1] != "f": # type: ignore
3672 x = int(round(x))
3673 return x
3675 def m2i(self, pkt, x):
3676 # type: (Optional[Packet], Union[int, float]) -> Union[int, float]
3677 x = x * self.scaling + self.offset
3678 if isinstance(x, float) and self.fmt[-1] != "f": # type: ignore
3679 x = round(x, self.ndigits)
3680 return x
3682 def any2i(self, pkt, x):
3683 # type: (Optional[Packet], Any) -> Union[int, float, None]
3684 if isinstance(x, (str, bytes)):
3685 x = struct.unpack(self.fmt, bytes_encode(x))[0] # type: ignore
3686 x = self.m2i(pkt, x)
3687 if not isinstance(x, (int, float)) and x is not None:
3688 raise ValueError("Unknown type")
3689 return x
3691 def i2repr(self, pkt, x):
3692 # type: (Optional[Packet], Union[int, float]) -> str
3693 return "%s %s" % (
3694 self.i2h(pkt, x), # type: ignore
3695 self.unit
3696 )
3698 def randval(self):
3699 # type: () -> RandFloat
3700 value = Field.randval(self) # type: ignore
3701 if value is not None:
3702 min_val = round(value.min * self.scaling + self.offset,
3703 self.ndigits)
3704 max_val = round(value.max * self.scaling + self.offset,
3705 self.ndigits)
3707 return RandFloat(min(min_val, max_val), max(min_val, max_val))
3710class ScalingField(_ScalingField,
3711 Field[Union[int, float], Union[int, float]]):
3712 """ Handle physical values which are scaled and/or offset for communication
3714 Example:
3715 >>> from scapy.packet import Packet
3716 >>> class ScalingFieldTest(Packet):
3717 fields_desc = [ScalingField('data', 0, scaling=0.1, offset=-1, unit='mV')] # noqa: E501
3718 >>> ScalingFieldTest(data=10).show2()
3719 ###[ ScalingFieldTest ]###
3720 data= 10.0 mV
3721 >>> hexdump(ScalingFieldTest(data=10))
3722 0000 6E n
3723 >>> hexdump(ScalingFieldTest(data=b"\x6D"))
3724 0000 6D m
3725 >>> ScalingFieldTest(data=b"\x6D").show2()
3726 ###[ ScalingFieldTest ]###
3727 data= 9.9 mV
3729 bytes(ScalingFieldTest(...)) will produce 0x6E in this example.
3730 0x6E is 110 (decimal). This is calculated through the scaling factor
3731 and the offset. "data" was set to 10, which means, we want to transfer
3732 the physical value 10 mV. To calculate the value, which has to be
3733 sent on the bus, the offset has to subtracted and the scaling has to be
3734 applied by division through the scaling factor.
3735 bytes = (data - offset) / scaling
3736 bytes = ( 10 - (-1) ) / 0.1
3737 bytes = 110 = 0x6E
3739 If you want to force a certain internal value, you can assign a byte-
3740 string to the field (data=b"\x6D"). If a string of a bytes object is
3741 given to the field, no internal value conversion will be applied
3743 :param name: field's name
3744 :param default: default value for the field
3745 :param scaling: scaling factor for the internal value conversion
3746 :param unit: string for the unit representation of the internal value
3747 :param offset: value to offset the internal value during conversion
3748 :param ndigits: number of fractional digits for the internal conversion
3749 :param fmt: struct.pack format used to parse and serialize the internal value from and to machine representation # noqa: E501
3750 """
3753class BitScalingField(_ScalingField, BitField): # type: ignore
3754 """
3755 A ScalingField that is a BitField
3756 """
3758 def __init__(self, name, default, size, *args, **kwargs):
3759 # type: (str, int, int, *Any, **Any) -> None
3760 _ScalingField.__init__(self, name, default, *args, **kwargs)
3761 BitField.__init__(self, name, default, size) # type: ignore
3764class OUIField(X3BytesField):
3765 """
3766 A field designed to carry a OUI (3 bytes)
3767 """
3769 def i2repr(self, pkt, val):
3770 # type: (Optional[Packet], int) -> str
3771 by_val = struct.pack("!I", val or 0)[1:]
3772 oui = str2mac(by_val + b"\0" * 3)[:8]
3773 if conf.manufdb:
3774 fancy = conf.manufdb._get_manuf(oui)
3775 if fancy != oui:
3776 return "%s (%s)" % (fancy, oui)
3777 return oui
3780class UUIDField(Field[UUID, bytes]):
3781 """Field for UUID storage, wrapping Python's uuid.UUID type.
3783 The internal storage format of this field is ``uuid.UUID`` from the Python
3784 standard library.
3786 There are three formats (``uuid_fmt``) for this field type:
3788 * ``FORMAT_BE`` (default): the UUID is six fields in big-endian byte order,
3789 per RFC 4122.
3791 This format is used by DHCPv6 (RFC 6355) and most network protocols.
3793 * ``FORMAT_LE``: the UUID is six fields, with ``time_low``, ``time_mid``
3794 and ``time_high_version`` in little-endian byte order. This *doesn't*
3795 change the arrangement of the fields from RFC 4122.
3797 This format is used by Microsoft's COM/OLE libraries.
3799 * ``FORMAT_REV``: the UUID is a single 128-bit integer in little-endian
3800 byte order. This *changes the arrangement* of the fields.
3802 This format is used by Bluetooth Low Energy.
3804 Note: You should use the constants here.
3806 The "human encoding" of this field supports a number of different input
3807 formats, and wraps Python's ``uuid.UUID`` library appropriately:
3809 * Given a bytearray, bytes or str of 16 bytes, this class decodes UUIDs in
3810 wire format.
3812 * Given a bytearray, bytes or str of other lengths, this delegates to
3813 ``uuid.UUID`` the Python standard library. This supports a number of
3814 different encoding options -- see the Python standard library
3815 documentation for more details.
3817 * Given an int or long, presumed to be a 128-bit integer to pass to
3818 ``uuid.UUID``.
3820 * Given a tuple:
3822 * Tuples of 11 integers are treated as having the last 6 integers forming
3823 the ``node`` field, and are merged before being passed as a tuple of 6
3824 integers to ``uuid.UUID``.
3826 * Otherwise, the tuple is passed as the ``fields`` parameter to
3827 ``uuid.UUID`` directly without modification.
3829 ``uuid.UUID`` expects a tuple of 6 integers.
3831 Other types (such as ``uuid.UUID``) are passed through.
3832 """
3834 __slots__ = ["uuid_fmt"]
3836 FORMAT_BE = 0
3837 FORMAT_LE = 1
3838 FORMAT_REV = 2
3840 # Change this when we get new formats
3841 FORMATS = (FORMAT_BE, FORMAT_LE, FORMAT_REV)
3843 def __init__(self, name, default, uuid_fmt=FORMAT_BE):
3844 # type: (str, Optional[int], int) -> None
3845 self.uuid_fmt = uuid_fmt
3846 self._check_uuid_fmt()
3847 Field.__init__(self, name, default, "16s")
3849 def _check_uuid_fmt(self):
3850 # type: () -> None
3851 """Checks .uuid_fmt, and raises an exception if it is not valid."""
3852 if self.uuid_fmt not in UUIDField.FORMATS:
3853 raise FieldValueRangeException(
3854 "Unsupported uuid_fmt ({})".format(self.uuid_fmt))
3856 def i2m(self, pkt, x):
3857 # type: (Optional[Packet], Optional[UUID]) -> bytes
3858 self._check_uuid_fmt()
3859 if x is None:
3860 return b'\0' * 16
3861 if self.uuid_fmt == UUIDField.FORMAT_BE:
3862 return x.bytes
3863 elif self.uuid_fmt == UUIDField.FORMAT_LE:
3864 return x.bytes_le
3865 elif self.uuid_fmt == UUIDField.FORMAT_REV:
3866 return x.bytes[::-1]
3867 else:
3868 raise FieldAttributeException("Unknown fmt")
3870 def m2i(self,
3871 pkt, # type: Optional[Packet]
3872 x, # type: bytes
3873 ):
3874 # type: (...) -> UUID
3875 self._check_uuid_fmt()
3876 if self.uuid_fmt == UUIDField.FORMAT_BE:
3877 return UUID(bytes=x)
3878 elif self.uuid_fmt == UUIDField.FORMAT_LE:
3879 return UUID(bytes_le=x)
3880 elif self.uuid_fmt == UUIDField.FORMAT_REV:
3881 return UUID(bytes=x[::-1])
3882 else:
3883 raise FieldAttributeException("Unknown fmt")
3885 def any2i(self,
3886 pkt, # type: Optional[Packet]
3887 x # type: Any # noqa: E501
3888 ):
3889 # type: (...) -> Optional[UUID]
3890 # Python's uuid doesn't handle bytearray, so convert to an immutable
3891 # type first.
3892 if isinstance(x, bytearray):
3893 x = bytes_encode(x)
3895 if isinstance(x, int):
3896 u = UUID(int=x)
3897 elif isinstance(x, tuple):
3898 if len(x) == 11:
3899 # For compatibility with dce_rpc: this packs into a tuple where
3900 # elements 7..10 are the 48-bit node ID.
3901 node = 0
3902 for i in x[5:]:
3903 node = (node << 8) | i
3905 x = (x[0], x[1], x[2], x[3], x[4], node)
3907 u = UUID(fields=x)
3908 elif isinstance(x, (str, bytes)):
3909 if len(x) == 16:
3910 # Raw bytes
3911 u = self.m2i(pkt, bytes_encode(x))
3912 else:
3913 u = UUID(plain_str(x))
3914 elif isinstance(x, (UUID, RandUUID)):
3915 u = cast(UUID, x)
3916 else:
3917 return None
3918 return u
3920 @staticmethod
3921 def randval():
3922 # type: () -> RandUUID
3923 return RandUUID()
3926class UUIDEnumField(UUIDField, _EnumField[UUID]):
3927 __slots__ = EnumField.__slots__
3929 def __init__(self, name, default, enum, uuid_fmt=0):
3930 # type: (str, Optional[int], Any, int) -> None
3931 _EnumField.__init__(self, name, default, enum, "16s") # type: ignore
3932 UUIDField.__init__(self, name, default, uuid_fmt=uuid_fmt)
3934 def any2i(self, pkt, x):
3935 # type: (Optional[Packet], Any) -> UUID
3936 return _EnumField.any2i(self, pkt, x) # type: ignore
3938 def i2repr(self,
3939 pkt, # type: Optional[Packet]
3940 x, # type: UUID
3941 ):
3942 # type: (...) -> Any
3943 return _EnumField.i2repr(self, pkt, x)
3946class BitExtendedField(Field[Optional[int], int]):
3947 """
3948 Low E Bit Extended Field
3950 This type of field has a variable number of bytes. Each byte is defined
3951 as follows:
3952 - 7 bits of data
3953 - 1 bit an an extension bit:
3955 + 0 means it is last byte of the field ("stopping bit")
3956 + 1 means there is another byte after this one ("forwarding bit")
3958 To get the actual data, it is necessary to hop the binary data byte per
3959 byte and to check the extension bit until 0
3960 """
3962 __slots__ = ["extension_bit"]
3964 def __init__(self, name, default, extension_bit):
3965 # type: (str, Optional[Any], int) -> None
3966 Field.__init__(self, name, default, "B")
3967 assert extension_bit in [7, 0]
3968 self.extension_bit = extension_bit
3970 def addfield(self, pkt, s, val):
3971 # type: (Optional[Packet], bytes, Optional[int]) -> bytes
3972 val = self.i2m(pkt, val)
3973 if not val:
3974 return s + b"\0"
3975 rv = b""
3976 mask = 1 << self.extension_bit
3977 shift = (self.extension_bit + 1) % 8
3978 while val:
3979 bv = (val & 0x7F) << shift
3980 val = val >> 7
3981 if val:
3982 bv |= mask
3983 rv += struct.pack("!B", bv)
3984 return s + rv
3986 def getfield(self, pkt, s):
3987 # type: (Optional[Any], bytes) -> Tuple[bytes, Optional[int]]
3988 val = 0
3989 smask = 1 << self.extension_bit
3990 mask = 0xFF & ~ (1 << self.extension_bit)
3991 shift = (self.extension_bit + 1) % 8
3992 i = 0
3993 while s:
3994 val |= ((s[0] & mask) >> shift) << (7 * i)
3995 if (s[0] & smask) == 0: # extension bit is 0
3996 # end
3997 s = s[1:]
3998 break
3999 s = s[1:]
4000 i += 1
4001 return s, self.m2i(pkt, val)
4004class LSBExtendedField(BitExtendedField):
4005 # This is a BitExtendedField with the extension bit on LSB
4006 def __init__(self, name, default):
4007 # type: (str, Optional[Any]) -> None
4008 BitExtendedField.__init__(self, name, default, extension_bit=0)
4011class MSBExtendedField(BitExtendedField):
4012 # This is a BitExtendedField with the extension bit on MSB
4013 def __init__(self, name, default):
4014 # type: (str, Optional[Any]) -> None
4015 BitExtendedField.__init__(self, name, default, extension_bit=7)