| ## This file is part of Scapy |
| ## See http://www.secdev.org/projects/scapy for more informations |
| ## Copyright (C) Philippe Biondi <phil@secdev.org> |
| ## This program is published under a GPLv2 license |
| |
| """ |
| Fields: basic data structures that make up parts of packets. |
| """ |
| |
| from __future__ import absolute_import |
| import struct,copy,socket,collections |
| from scapy.config import conf |
| from scapy.dadict import DADict |
| from scapy.volatile import * |
| from scapy.data import * |
| from scapy.compat import * |
| from scapy.utils import * |
| from scapy.base_classes import BasePacket, Gen, Net, Field_metaclass |
| from scapy.error import warning |
| import scapy.modules.six as six |
| from scapy.modules.six.moves import range |
| |
| |
| ############ |
| ## Fields ## |
| ############ |
| |
| class Field(six.with_metaclass(Field_metaclass, object)): |
| """For more informations on how this work, please refer to |
| http://www.secdev.org/projects/scapy/files/scapydoc.pdf |
| chapter ``Adding a New Field''""" |
| __slots__ = ["name", "fmt", "default", "sz", "owners"] |
| islist = 0 |
| ismutable = False |
| holds_packets = 0 |
| def __init__(self, name, default, fmt="H"): |
| self.name = name |
| if fmt[0] in "@=<>!": |
| self.fmt = fmt |
| else: |
| self.fmt = "!"+fmt |
| self.default = self.any2i(None,default) |
| self.sz = struct.calcsize(self.fmt) |
| self.owners = [] |
| |
| def register_owner(self, cls): |
| self.owners.append(cls) |
| |
| def i2len(self, pkt, x): |
| """Convert internal value to a length usable by a FieldLenField""" |
| return self.sz |
| def i2count(self, pkt, x): |
| """Convert internal value to a number of elements usable by a FieldLenField. |
| Always 1 except for list fields""" |
| return 1 |
| def h2i(self, pkt, x): |
| """Convert human value to internal value""" |
| return x |
| def i2h(self, pkt, x): |
| """Convert internal value to human value""" |
| return x |
| def m2i(self, pkt, x): |
| """Convert machine value to internal value""" |
| return x |
| def i2m(self, pkt, x): |
| """Convert internal value to machine value""" |
| if x is None: |
| x = 0 |
| elif isinstance(x, str): |
| return raw(x) |
| return x |
| def any2i(self, pkt, x): |
| """Try to understand the most input values possible and make an internal value from them""" |
| return self.h2i(pkt, x) |
| def i2repr(self, pkt, x): |
| """Convert internal value to a nice representation""" |
| return repr(self.i2h(pkt,x)) |
| def addfield(self, pkt, s, val): |
| """Add an internal value to a string""" |
| return s+struct.pack(self.fmt, self.i2m(pkt,val)) |
| def getfield(self, pkt, s): |
| """Extract an internal value from a string""" |
| return s[self.sz:], self.m2i(pkt, struct.unpack(self.fmt, s[:self.sz])[0]) |
| def do_copy(self, x): |
| if hasattr(x, "copy"): |
| return x.copy() |
| if isinstance(x, list): |
| x = x[:] |
| for i in range(len(x)): |
| if isinstance(x[i], BasePacket): |
| x[i] = x[i].copy() |
| return x |
| def __repr__(self): |
| return "<Field (%s).%s>" % (",".join(x.__name__ for x in self.owners),self.name) |
| def copy(self): |
| return copy.deepcopy(self) |
| def randval(self): |
| """Return a volatile object whose value is both random and suitable for this field""" |
| fmtt = self.fmt[-1] |
| if fmtt in "BHIQ": |
| return {"B":RandByte,"H":RandShort,"I":RandInt, "Q":RandLong}[fmtt]() |
| elif fmtt == "s": |
| if self.fmt[0] in "0123456789": |
| l = int(self.fmt[:-1]) |
| else: |
| l = int(self.fmt[1:-1]) |
| return RandBin(l) |
| else: |
| warning("no random class for [%s] (fmt=%s).", self.name, self.fmt) |
| |
| |
| |
| |
| class Emph(object): |
| __slots__ = ["fld"] |
| def __init__(self, fld): |
| self.fld = fld |
| def __getattr__(self, attr): |
| return getattr(self.fld,attr) |
| def __hash__(self): |
| return hash(self.fld) |
| def __eq__(self, other): |
| return self.fld == other |
| |
| |
| class ActionField(object): |
| __slots__ = ["_fld", "_action_method", "_privdata"] |
| def __init__(self, fld, action_method, **kargs): |
| self._fld = fld |
| self._action_method = action_method |
| self._privdata = kargs |
| def any2i(self, pkt, val): |
| getattr(pkt, self._action_method)(val, self._fld, **self._privdata) |
| return getattr(self._fld, "any2i")(pkt, val) |
| def __getattr__(self, attr): |
| return getattr(self._fld,attr) |
| |
| |
| class ConditionalField(object): |
| __slots__ = ["fld", "cond"] |
| def __init__(self, fld, cond): |
| self.fld = fld |
| self.cond = cond |
| def _evalcond(self,pkt): |
| return self.cond(pkt) |
| |
| def getfield(self, pkt, s): |
| if self._evalcond(pkt): |
| return self.fld.getfield(pkt,s) |
| else: |
| return s,None |
| |
| def addfield(self, pkt, s, val): |
| if self._evalcond(pkt): |
| return self.fld.addfield(pkt,s,val) |
| else: |
| return s |
| def __getattr__(self, attr): |
| return getattr(self.fld,attr) |
| |
| |
| class PadField(object): |
| """Add bytes after the proxified field so that it ends at the specified |
| alignment from its beginning""" |
| __slots__ = ["_fld", "_align", "_padwith"] |
| def __init__(self, fld, align, padwith=None): |
| self._fld = fld |
| self._align = align |
| self._padwith = padwith or b"" |
| |
| def padlen(self, flen): |
| return -flen%self._align |
| |
| def getfield(self, pkt, s): |
| remain,val = self._fld.getfield(pkt,s) |
| padlen = self.padlen(len(s)-len(remain)) |
| return remain[padlen:], val |
| |
| def addfield(self, pkt, s, val): |
| sval = self._fld.addfield(pkt, b"", val) |
| return s+sval+struct.pack("%is" % (self.padlen(len(sval))), self._padwith) |
| |
| def __getattr__(self, attr): |
| return getattr(self._fld,attr) |
| |
| |
| class DestField(Field): |
| __slots__ = ["defaultdst"] |
| # Each subclass must have its own bindings attribute |
| # bindings = {} |
| def __init__(self, name, default): |
| self.defaultdst = default |
| def dst_from_pkt(self, pkt): |
| for addr, condition in self.bindings.get(pkt.payload.__class__, []): |
| try: |
| if all(pkt.payload.getfieldval(field) == value |
| for field, value in six.iteritems(condition)): |
| return addr |
| except AttributeError: |
| pass |
| return self.defaultdst |
| @classmethod |
| def bind_addr(cls, layer, addr, **condition): |
| cls.bindings.setdefault(layer, []).append((addr, condition)) |
| |
| |
| class MACField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "6s") |
| def i2m(self, pkt, x): |
| if x is None: |
| return b"\0\0\0\0\0\0" |
| return mac2str(x) |
| def m2i(self, pkt, x): |
| return str2mac(x) |
| def any2i(self, pkt, x): |
| if isinstance(x, bytes) and len(x) == 6: |
| x = self.m2i(pkt, x) |
| return x |
| def i2repr(self, pkt, x): |
| x = self.i2h(pkt, x) |
| if self in conf.resolve: |
| x = conf.manufdb._resolve_MAC(x) |
| return x |
| def randval(self): |
| return RandMAC() |
| |
| |
| class IPField(Field): |
| slots = [] |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "4s") |
| def h2i(self, pkt, x): |
| if isinstance(x, bytes): |
| x = plain_str(x) |
| if isinstance(x, str): |
| try: |
| inet_aton(x) |
| except socket.error: |
| x = Net(x) |
| elif isinstance(x, list): |
| x = [self.h2i(pkt, n) for n in x] |
| return x |
| def resolve(self, x): |
| if self in conf.resolve: |
| try: |
| ret = socket.gethostbyaddr(x)[0] |
| except: |
| pass |
| else: |
| if ret: |
| return ret |
| return x |
| def i2m(self, pkt, x): |
| return inet_aton(x) |
| def m2i(self, pkt, x): |
| return inet_ntoa(x) |
| def any2i(self, pkt, x): |
| return self.h2i(pkt,x) |
| def i2repr(self, pkt, x): |
| return self.resolve(self.i2h(pkt, x)) |
| def randval(self): |
| return RandIP() |
| |
| class SourceIPField(IPField): |
| __slots__ = ["dstname"] |
| def __init__(self, name, dstname): |
| IPField.__init__(self, name, None) |
| self.dstname = dstname |
| def __findaddr(self, pkt): |
| if conf.route is None: |
| # unused import, only to initialize conf.route |
| import scapy.route |
| dst = ("0.0.0.0" if self.dstname is None |
| else getattr(pkt, self.dstname)) |
| if isinstance(dst, (Gen, list)): |
| r = {conf.route.route(daddr) for daddr in dst} |
| if len(r) > 1: |
| warning("More than one possible route for %r" % (dst,)) |
| return min(r)[1] |
| return conf.route.route(dst)[1] |
| def i2m(self, pkt, x): |
| if x is None: |
| x = self.__findaddr(pkt) |
| return IPField.i2m(self, pkt, x) |
| def i2h(self, pkt, x): |
| if x is None: |
| x = self.__findaddr(pkt) |
| return IPField.i2h(self, pkt, x) |
| |
| |
| |
| |
| class ByteField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "B") |
| |
| class XByteField(ByteField): |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt, x)) |
| |
| class OByteField(ByteField): |
| def i2repr(self, pkt, x): |
| return "%03o"%self.i2h(pkt, x) |
| |
| class X3BytesField(XByteField): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "!I") |
| def addfield(self, pkt, s, val): |
| return s+struct.pack(self.fmt, self.i2m(pkt,val))[1:4] |
| def getfield(self, pkt, s): |
| return s[3:], self.m2i(pkt, struct.unpack(self.fmt, b"\x00"+s[:3])[0]) |
| |
| class ThreeBytesField(X3BytesField, ByteField): |
| def i2repr(self, pkt, x): |
| return ByteField.i2repr(self, pkt, x) |
| |
| class SignedByteField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "b") |
| |
| class ShortField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "H") |
| |
| class SignedShortField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "h") |
| |
| class LEShortField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "<H") |
| |
| class XShortField(ShortField): |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt, x)) |
| |
| |
| class IntField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "I") |
| |
| class SignedIntField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "i") |
| def randval(self): |
| return RandSInt() |
| |
| class LEIntField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "<I") |
| |
| class LESignedIntField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "<i") |
| def randval(self): |
| return RandSInt() |
| |
| class XIntField(IntField): |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt, x)) |
| |
| |
| class LongField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "Q") |
| |
| class LELongField(LongField): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "<Q") |
| |
| class XLongField(LongField): |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt, x)) |
| |
| class IEEEFloatField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "f") |
| |
| class IEEEDoubleField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "d") |
| |
| |
| class StrField(Field): |
| __slots__ = ["remain"] |
| def __init__(self, name, default, fmt="H", remain=0): |
| Field.__init__(self,name,default,fmt) |
| self.remain = remain |
| def i2len(self, pkt, i): |
| return len(i) |
| def any2i(self, pkt, x): |
| if isinstance(x, str if six.PY3 else unicode): |
| x = raw(x) |
| return super(StrField, self).any2i(pkt, x) |
| def i2repr(self, pkt, x): |
| val = super(StrField, self).i2repr(pkt, x) |
| if val[:2] in ['b"', "b'"]: |
| return val[1:] |
| return val |
| def i2m(self, pkt, x): |
| if x is None: |
| return b"" |
| if not isinstance(x, bytes): |
| return raw(x) |
| return x |
| def addfield(self, pkt, s, val): |
| return s + self.i2m(pkt, val) |
| def getfield(self, pkt, s): |
| if self.remain == 0: |
| return b"", self.m2i(pkt, s) |
| else: |
| return s[-self.remain:],self.m2i(pkt, s[:-self.remain]) |
| def randval(self): |
| return RandBin(RandNum(0,1200)) |
| |
| class PacketField(StrField): |
| __slots__ = ["cls"] |
| holds_packets = 1 |
| def __init__(self, name, default, cls, remain=0): |
| StrField.__init__(self, name, default, remain=remain) |
| self.cls = cls |
| def i2m(self, pkt, i): |
| if i is None: |
| return b"" |
| return raw(i) |
| def m2i(self, pkt, m): |
| return self.cls(m) |
| def getfield(self, pkt, s): |
| i = self.m2i(pkt, s) |
| remain = b"" |
| if conf.padding_layer in i: |
| r = i[conf.padding_layer] |
| del(r.underlayer.payload) |
| remain = r.load |
| return remain,i |
| |
| class PacketLenField(PacketField): |
| __slots__ = ["length_from"] |
| def __init__(self, name, default, cls, length_from=None): |
| PacketField.__init__(self, name, default, cls) |
| self.length_from = length_from |
| def getfield(self, pkt, s): |
| l = self.length_from(pkt) |
| try: |
| i = self.m2i(pkt, s[:l]) |
| except Exception: |
| if conf.debug_dissector: |
| raise |
| i = conf.raw_layer(load=s[:l]) |
| return s[l:],i |
| |
| |
| class PacketListField(PacketField): |
| """ PacketListField represents a series of Packet instances that might occur right in the middle of another Packet |
| field list. |
| This field type may also be used to indicate that a series of Packet instances have a sibling semantic instead of |
| a parent/child relationship (i.e. a stack of layers). |
| """ |
| __slots__ = ["count_from", "length_from", "next_cls_cb"] |
| islist = 1 |
| def __init__(self, name, default, cls=None, count_from=None, length_from=None, next_cls_cb=None): |
| """ The number of Packet instances that are dissected by this field can be parametrized using one of three |
| different mechanisms/parameters: |
| * count_from: a callback that returns the number of Packet instances to dissect. The callback prototype is: |
| count_from(pkt:Packet) -> int |
| * length_from: a callback that returns the number of bytes that must be dissected by this field. The |
| callback prototype is: |
| length_from(pkt:Packet) -> int |
| * next_cls_cb: a callback that enables a Scapy developer to dynamically discover if another Packet instance |
| should be dissected or not. See below for this callback prototype. |
| |
| The bytes that are not consumed during the dissection of this field are passed to the next field of the current |
| packet. |
| |
| For the serialization of such a field, the list of Packets that are contained in a PacketListField can be |
| heterogeneous and is unrestricted. |
| |
| The type of the Packet instances that are dissected with this field is specified or discovered using one of the |
| following mechanism: |
| * the cls parameter may contain a callable that returns an instance of the dissected Packet. This |
| may either be a reference of a Packet subclass (e.g. DNSRROPT in layers/dns.py) to generate an |
| homogeneous PacketListField or a function deciding the type of the Packet instance |
| (e.g. _CDPGuessAddrRecord in contrib/cdp.py) |
| * the cls parameter may contain a class object with a defined "dispatch_hook" classmethod. That |
| method must return a Packet instance. The dispatch_hook callmethod must implement the following prototype: |
| dispatch_hook(cls, _pkt:Optional[Packet], *args, **kargs) -> Packet_metaclass |
| The _pkt parameter may contain a reference to the packet instance containing the PacketListField that is |
| being dissected. |
| * the next_cls_cb parameter may contain a callable whose prototype is: |
| cbk(pkt:Packet, lst:List[Packet], cur:Optional[Packet], remain:str) -> Optional[Packet_metaclass] |
| The pkt argument contains a reference to the Packet instance containing the PacketListField that is |
| being dissected. The lst argument is the list of all Packet instances that were previously parsed during |
| the current PacketListField dissection, save for the very last Packet instance. The cur argument |
| contains a reference to that very last parsed Packet instance. The remain argument contains the bytes |
| that may still be consumed by the current PacketListField dissection operation. This callback returns |
| either the type of the next Packet to dissect or None to indicate that no more Packet are to be |
| dissected. |
| These four arguments allows a variety of dynamic discovery of the number of Packet to dissect and of the |
| type of each one of these Packets, including: type determination based on current Packet instances or |
| its underlayers, continuation based on the previously parsed Packet instances within that |
| PacketListField, continuation based on a look-ahead on the bytes to be dissected... |
| |
| The cls and next_cls_cb parameters are semantically exclusive, although one could specify both. If both are |
| specified, cls is silently ignored. The same is true for count_from and next_cls_cb. |
| length_from and next_cls_cb are compatible and the dissection will end, whichever of the two stop conditions |
| comes first. |
| |
| @param name: the name of the field |
| @param default: the default value of this field; generally an empty Python list |
| @param cls: either a callable returning a Packet instance or a class object defining a dispatch_hook class |
| method |
| @param count_from: a callback returning the number of Packet instances to dissect |
| @param length_from: a callback returning the number of bytes to dissect |
| @param next_cls_cb: a callback returning either None or the type of the next Packet to dissect. |
| """ |
| if default is None: |
| default = [] # Create a new list for each instance |
| PacketField.__init__(self, name, default, cls) |
| self.count_from = count_from |
| self.length_from = length_from |
| self.next_cls_cb = next_cls_cb |
| |
| def any2i(self, pkt, x): |
| if not isinstance(x, list): |
| return [x] |
| else: |
| return x |
| def i2count(self, pkt, val): |
| if isinstance(val, list): |
| return len(val) |
| return 1 |
| def i2len(self, pkt, val): |
| return sum( len(p) for p in val ) |
| def do_copy(self, x): |
| if x is None: |
| return None |
| else: |
| return [p if isinstance(p, six.string_types) else p.copy() for p in x] |
| def getfield(self, pkt, s): |
| c = l = cls = None |
| if self.length_from is not None: |
| l = self.length_from(pkt) |
| elif self.count_from is not None: |
| c = self.count_from(pkt) |
| if self.next_cls_cb is not None: |
| cls = self.next_cls_cb(pkt, [], None, s) |
| c = 1 |
| |
| lst = [] |
| ret = b"" |
| remain = s |
| if l is not None: |
| remain,ret = s[:l],s[l:] |
| while remain: |
| if c is not None: |
| if c <= 0: |
| break |
| c -= 1 |
| try: |
| if cls is not None: |
| p = cls(remain) |
| else: |
| p = self.m2i(pkt, remain) |
| except Exception: |
| if conf.debug_dissector: |
| raise |
| p = conf.raw_layer(load=remain) |
| remain = b"" |
| else: |
| if conf.padding_layer in p: |
| pad = p[conf.padding_layer] |
| remain = pad.load |
| del(pad.underlayer.payload) |
| if self.next_cls_cb is not None: |
| cls = self.next_cls_cb(pkt, lst, p, remain) |
| if cls is not None: |
| c += 1 |
| else: |
| remain = b"" |
| lst.append(p) |
| return remain+ret,lst |
| def addfield(self, pkt, s, val): |
| return s + b"".join(raw(v) for v in val) |
| |
| |
| class StrFixedLenField(StrField): |
| __slots__ = ["length_from"] |
| def __init__(self, name, default, length=None, length_from=None): |
| StrField.__init__(self, name, default) |
| self.length_from = length_from |
| if length is not None: |
| self.length_from = lambda pkt,length=length: length |
| def i2repr(self, pkt, v): |
| if isinstance(v, bytes): |
| v = v.rstrip(b"\0") |
| return super(StrFixedLenField, self).i2repr(pkt, v) |
| def getfield(self, pkt, s): |
| l = self.length_from(pkt) |
| return s[l:], self.m2i(pkt,s[:l]) |
| def addfield(self, pkt, s, val): |
| l = self.length_from(pkt) |
| return s+struct.pack("%is"%l,self.i2m(pkt, val)) |
| def randval(self): |
| try: |
| l = self.length_from(None) |
| except: |
| l = RandNum(0,200) |
| return RandBin(l) |
| |
| class StrFixedLenEnumField(StrFixedLenField): |
| __slots__ = ["enum"] |
| def __init__(self, name, default, length=None, enum=None, length_from=None): |
| StrFixedLenField.__init__(self, name, default, length=length, length_from=length_from) |
| self.enum = enum |
| def i2repr(self, pkt, v): |
| r = v.rstrip("\0") |
| rr = repr(r) |
| if v in self.enum: |
| rr = "%s (%s)" % (rr, self.enum[v]) |
| elif r in self.enum: |
| rr = "%s (%s)" % (rr, self.enum[r]) |
| return rr |
| |
| class NetBIOSNameField(StrFixedLenField): |
| def __init__(self, name, default, length=31): |
| StrFixedLenField.__init__(self, name, default, length) |
| def i2m(self, pkt, x): |
| l = self.length_from(pkt)//2 |
| if x is None: |
| x = b"" |
| x += b" "*(l) |
| x = x[:l] |
| x = b"".join(chb(0x41 + orb(b)>>4) + chb(0x41 + orb(b)&0xf) for b in x) |
| x = b" "+x |
| return x |
| def m2i(self, pkt, x): |
| x = x.strip(b"\x00").strip(b" ") |
| return b"".join(map(lambda x,y: chb((((orb(x)-1)&0xf)<<4)+((orb(y)-1)&0xf)), x[::2],x[1::2])) |
| |
| class StrLenField(StrField): |
| __slots__ = ["length_from"] |
| def __init__(self, name, default, fld=None, length_from=None): |
| StrField.__init__(self, name, default) |
| self.length_from = length_from |
| def getfield(self, pkt, s): |
| l = self.length_from(pkt) |
| return s[l:], self.m2i(pkt,s[:l]) |
| |
| class XStrField(StrField): |
| """ |
| StrField which value is printed as hexadecimal. |
| """ |
| |
| def i2repr(self, pkt, x): |
| if x is None: |
| return repr(x) |
| return bytes_hex(x).decode() |
| |
| class XStrLenField(StrLenField): |
| """ |
| StrLenField which value is printed as hexadecimal. |
| """ |
| |
| def i2repr(self, pkt, x): |
| if not x: |
| return repr(x) |
| return bytes_hex(x[:self.length_from(pkt)]).decode() |
| |
| class XStrFixedLenField(StrFixedLenField): |
| """ |
| StrFixedLenField which value is printed as hexadecimal. |
| """ |
| |
| def i2repr(self, pkt, x): |
| if not x: |
| return repr(x) |
| return bytes_hex(x[:self.length_from(pkt)]).decode() |
| |
| class StrLenFieldUtf16(StrLenField): |
| def h2i(self, pkt, x): |
| return plain_str(x).encode('utf-16')[2:] |
| def i2h(self, pkt, x): |
| return x.decode('utf-16') |
| |
| class BoundStrLenField(StrLenField): |
| __slots__ = ["minlen", "maxlen"] |
| def __init__(self,name, default, minlen= 0, maxlen= 255, fld=None, length_from=None): |
| StrLenField.__init__(self, name, default, fld, length_from) |
| self.minlen = minlen |
| self.maxlen = maxlen |
| |
| def randval(self): |
| return RandBin(RandNum(self.minlen, self.maxlen)) |
| |
| class FieldListField(Field): |
| __slots__ = ["field", "count_from", "length_from"] |
| islist = 1 |
| def __init__(self, name, default, field, length_from=None, count_from=None): |
| if default is None: |
| default = [] # Create a new list for each instance |
| self.field = field |
| Field.__init__(self, name, default) |
| self.count_from = count_from |
| self.length_from = length_from |
| |
| def i2count(self, pkt, val): |
| if isinstance(val, list): |
| return len(val) |
| return 1 |
| def i2len(self, pkt, val): |
| return int(sum(self.field.i2len(pkt,v) for v in val)) |
| |
| def i2m(self, pkt, val): |
| if val is None: |
| val = [] |
| return val |
| def any2i(self, pkt, x): |
| if not isinstance(x, list): |
| return [self.field.any2i(pkt, x)] |
| else: |
| return [self.field.any2i(pkt, e) for e in x] |
| def i2repr(self, pkt, x): |
| res = [] |
| for v in x: |
| r = self.field.i2repr(pkt, v) |
| res.append(r) |
| return "[%s]" % ", ".join(res) |
| def addfield(self, pkt, s, val): |
| val = self.i2m(pkt, val) |
| for v in val: |
| s = self.field.addfield(pkt, s, v) |
| return s |
| def getfield(self, pkt, s): |
| c = l = None |
| if self.length_from is not None: |
| l = self.length_from(pkt) |
| elif self.count_from is not None: |
| c = self.count_from(pkt) |
| |
| val = [] |
| ret = b"" |
| if l is not None: |
| s,ret = s[:l],s[l:] |
| |
| while s: |
| if c is not None: |
| if c <= 0: |
| break |
| c -= 1 |
| s,v = self.field.getfield(pkt, s) |
| val.append(v) |
| return s+ret, val |
| |
| class FieldLenField(Field): |
| __slots__ = ["length_of", "count_of", "adjust"] |
| def __init__(self, name, default, length_of=None, fmt = "H", count_of=None, adjust=lambda pkt,x:x, fld=None): |
| Field.__init__(self, name, default, fmt) |
| self.length_of = length_of |
| self.count_of = count_of |
| self.adjust = adjust |
| if fld is not None: |
| #FIELD_LENGTH_MANAGEMENT_DEPRECATION(self.__class__.__name__) |
| self.length_of = fld |
| def i2m(self, pkt, x): |
| if x is None: |
| if self.length_of is not None: |
| fld,fval = pkt.getfield_and_val(self.length_of) |
| f = fld.i2len(pkt, fval) |
| else: |
| fld,fval = pkt.getfield_and_val(self.count_of) |
| f = fld.i2count(pkt, fval) |
| x = self.adjust(pkt,f) |
| return x |
| |
| class StrNullField(StrField): |
| def addfield(self, pkt, s, val): |
| return s+self.i2m(pkt, val)+b"\x00" |
| def getfield(self, pkt, s): |
| l = s.find(b"\x00") |
| if l < 0: |
| #XXX \x00 not found |
| return b"",s |
| return s[l+1:],self.m2i(pkt, s[:l]) |
| def randval(self): |
| return RandTermString(RandNum(0,1200),b"\x00") |
| |
| class StrStopField(StrField): |
| __slots__ = ["stop", "additionnal"] |
| def __init__(self, name, default, stop, additionnal=0): |
| Field.__init__(self, name, default) |
| self.stop = stop |
| self.additionnal = additionnal |
| def getfield(self, pkt, s): |
| l = s.find(self.stop) |
| if l < 0: |
| return b"",s |
| # raise Scapy_Exception,"StrStopField: stop value [%s] not found" %stop |
| l += len(self.stop)+self.additionnal |
| return s[l:],s[:l] |
| def randval(self): |
| return RandTermString(RandNum(0,1200),self.stop) |
| |
| class LenField(Field): |
| __slots__ = ["adjust"] |
| def __init__(self, name, default, fmt="H", adjust=lambda x: x): |
| Field.__init__(self, name, default, fmt) |
| self.adjust = adjust |
| def i2m(self, pkt, x): |
| if x is None: |
| x = self.adjust(len(pkt.payload)) |
| return x |
| |
| class BCDFloatField(Field): |
| def i2m(self, pkt, x): |
| return int(256*x) |
| def m2i(self, pkt, x): |
| return x/256.0 |
| |
| class BitField(Field): |
| __slots__ = ["rev", "size"] |
| def __init__(self, name, default, size): |
| Field.__init__(self, name, default) |
| self.rev = size < 0 |
| self.size = abs(size) |
| def reverse(self, val): |
| if self.size == 16: |
| # Replaces socket.ntohs (but work on both little/big endian) |
| val = struct.unpack('>H',struct.pack('<H', int(val)))[0] |
| elif self.size == 32: |
| # Same here but for socket.ntohl |
| val = struct.unpack('>I',struct.pack('<I', int(val)))[0] |
| return val |
| |
| def addfield(self, pkt, s, val): |
| val = self.i2m(pkt, val) |
| if isinstance(s, tuple): |
| s,bitsdone,v = s |
| else: |
| bitsdone = 0 |
| v = 0 |
| if self.rev: |
| val = self.reverse(val) |
| v <<= self.size |
| v |= val & ((1<<self.size) - 1) |
| bitsdone += self.size |
| while bitsdone >= 8: |
| bitsdone -= 8 |
| s = s+struct.pack("!B", v >> bitsdone) |
| v &= (1<<bitsdone)-1 |
| if bitsdone: |
| return s,bitsdone,v |
| else: |
| return s |
| def getfield(self, pkt, s): |
| if isinstance(s, tuple): |
| s,bn = s |
| else: |
| bn = 0 |
| # we don't want to process all the string |
| nb_bytes = (self.size+bn-1)//8 + 1 |
| w = s[:nb_bytes] |
| |
| # split the substring byte by byte |
| _bytes = struct.unpack('!%dB' % nb_bytes , w) |
| |
| b = 0 |
| for c in range(nb_bytes): |
| b |= int(_bytes[c]) << (nb_bytes-c-1)*8 |
| |
| # get rid of high order bits |
| b &= (1 << (nb_bytes*8-bn)) - 1 |
| |
| # remove low order bits |
| b = b >> (nb_bytes*8 - self.size - bn) |
| |
| if self.rev: |
| b = self.reverse(b) |
| |
| bn += self.size |
| s = s[bn//8:] |
| bn = bn%8 |
| b = self.m2i(pkt, b) |
| if bn: |
| return (s,bn),b |
| else: |
| return s,b |
| def randval(self): |
| return RandNum(0,2**self.size-1) |
| def i2len(self, pkt, x): |
| return float(self.size)/8 |
| |
| |
| class BitFieldLenField(BitField): |
| __slots__ = ["length_of", "count_of", "adjust"] |
| def __init__(self, name, default, size, length_of=None, count_of=None, adjust=lambda pkt,x:x): |
| BitField.__init__(self, name, default, size) |
| self.length_of = length_of |
| self.count_of = count_of |
| self.adjust = adjust |
| def i2m(self, pkt, x): |
| return (FieldLenField.i2m.__func__ if six.PY2 else FieldLenField.i2m)(self, pkt, x) |
| |
| |
| class XBitField(BitField): |
| def i2repr(self, pkt, x): |
| return lhex(self.i2h(pkt,x)) |
| |
| |
| class _EnumField(Field): |
| def __init__(self, name, default, enum, fmt = "H"): |
| """ Initializes enum fields. |
| |
| @param name: name of this field |
| @param default: default value of this field |
| @param enum: either a dict or a tuple of two callables. Dict keys are |
| the internal values, while the dict values are the |
| user-friendly representations. If the tuple is provided, |
| the first callable receives the internal value as |
| parameter and returns the user-friendly representation |
| and the second callable does the converse. The first |
| callable may return None to default to a literal string |
| (repr()) representation. |
| @param fmt: struct.pack format used to parse and serialize the |
| internal value from and to machine representation. |
| """ |
| if isinstance(enum, tuple): |
| self.i2s_cb = enum[0] |
| self.s2i_cb = enum[1] |
| self.i2s = None |
| self.s2i = None |
| else: |
| i2s = self.i2s = {} |
| s2i = self.s2i = {} |
| self.i2s_cb = None |
| self.s2i_cb = None |
| if isinstance(enum, list): |
| keys = range(len(enum)) |
| elif isinstance(enum, DADict): |
| keys = enum.iterkeys() |
| else: |
| keys = list(enum) |
| if any(isinstance(x, str) for x in keys): |
| i2s, s2i = s2i, i2s |
| for k in keys: |
| i2s[k] = enum[k] |
| s2i[enum[k]] = k |
| Field.__init__(self, name, default, fmt) |
| |
| def any2i_one(self, pkt, x): |
| if isinstance(x, str): |
| try: |
| x = self.s2i[x] |
| except TypeError: |
| x = self.s2i_cb(x) |
| return x |
| |
| def i2repr_one(self, pkt, x): |
| if self not in conf.noenum and not isinstance(x,VolatileValue): |
| try: |
| return self.i2s[x] |
| except KeyError: |
| pass |
| except TypeError: |
| ret = self.i2s_cb(x) |
| if ret is not None: |
| return ret |
| return repr(x) |
| |
| def any2i(self, pkt, x): |
| if isinstance(x, list): |
| return [self.any2i_one(pkt, z) for z in x] |
| else: |
| return self.any2i_one(pkt,x) |
| |
| def i2repr(self, pkt, x): |
| if isinstance(x, list): |
| return [self.i2repr_one(pkt, z) for z in x] |
| else: |
| return self.i2repr_one(pkt,x) |
| |
| class EnumField(_EnumField): |
| __slots__ = ["i2s", "s2i", "s2i_cb", "i2s_cb"] |
| |
| class CharEnumField(EnumField): |
| def __init__(self, name, default, enum, fmt = "1s"): |
| EnumField.__init__(self, name, default, enum, fmt) |
| if self.i2s is not None: |
| k = list(self.i2s) |
| if k and len(k[0]) != 1: |
| self.i2s,self.s2i = self.s2i,self.i2s |
| def any2i_one(self, pkt, x): |
| if len(x) != 1: |
| if self.s2i is None: |
| x = self.s2i_cb(x) |
| else: |
| x = self.s2i[x] |
| return x |
| |
| class BitEnumField(BitField, _EnumField): |
| __slots__ = EnumField.__slots__ |
| def __init__(self, name, default, size, enum): |
| _EnumField.__init__(self, name, default, enum) |
| self.rev = size < 0 |
| self.size = abs(size) |
| def any2i(self, pkt, x): |
| return _EnumField.any2i(self, pkt, x) |
| def i2repr(self, pkt, x): |
| return _EnumField.i2repr(self, pkt, x) |
| |
| class ShortEnumField(EnumField): |
| __slots__ = EnumField.__slots__ |
| def __init__(self, name, default, enum): |
| EnumField.__init__(self, name, default, enum, "H") |
| |
| class LEShortEnumField(EnumField): |
| def __init__(self, name, default, enum): |
| EnumField.__init__(self, name, default, enum, "<H") |
| |
| class ByteEnumField(EnumField): |
| def __init__(self, name, default, enum): |
| EnumField.__init__(self, name, default, enum, "B") |
| |
| class IntEnumField(EnumField): |
| def __init__(self, name, default, enum): |
| EnumField.__init__(self, name, default, enum, "I") |
| |
| class SignedIntEnumField(EnumField): |
| def __init__(self, name, default, enum): |
| EnumField.__init__(self, name, default, enum, "i") |
| def randval(self): |
| return RandSInt() |
| |
| class LEIntEnumField(EnumField): |
| def __init__(self, name, default, enum): |
| EnumField.__init__(self, name, default, enum, "<I") |
| |
| class XShortEnumField(ShortEnumField): |
| def i2repr_one(self, pkt, x): |
| if self not in conf.noenum and not isinstance(x,VolatileValue): |
| try: |
| return self.i2s[x] |
| except KeyError: |
| pass |
| except TypeError: |
| ret = self.i2s_cb(x) |
| if ret is not None: |
| return ret |
| return lhex(x) |
| |
| |
| class _MultiEnumField(_EnumField): |
| def __init__(self, name, default, enum, depends_on, fmt = "H"): |
| |
| self.depends_on = depends_on |
| self.i2s_multi = enum |
| self.s2i_multi = {} |
| self.s2i_all = {} |
| for m in enum: |
| self.s2i_multi[m] = s2i = {} |
| for k,v in six.iteritems(enum[m]): |
| s2i[v] = k |
| self.s2i_all[v] = k |
| Field.__init__(self, name, default, fmt) |
| def any2i_one(self, pkt, x): |
| if isinstance(x, str): |
| v = self.depends_on(pkt) |
| if v in self.s2i_multi: |
| s2i = self.s2i_multi[v] |
| if x in s2i: |
| return s2i[x] |
| return self.s2i_all[x] |
| return x |
| def i2repr_one(self, pkt, x): |
| v = self.depends_on(pkt) |
| if v in self.i2s_multi: |
| return self.i2s_multi[v].get(x,x) |
| return x |
| |
| class MultiEnumField(_MultiEnumField, EnumField): |
| __slots__ = ["depends_on", "i2s_multi", "s2i_multi", "s2i_all"] |
| |
| class BitMultiEnumField(BitField, _MultiEnumField): |
| __slots__ = EnumField.__slots__ + MultiEnumField.__slots__ |
| def __init__(self, name, default, size, enum, depends_on): |
| _MultiEnumField.__init__(self, name, default, enum, depends_on) |
| self.rev = size < 0 |
| self.size = abs(size) |
| def any2i(self, pkt, x): |
| return _MultiEnumField.any2i(self, pkt, x) |
| def i2repr(self, pkt, x): |
| return _MultiEnumField.i2repr(self, pkt, x) |
| |
| |
| class ByteEnumKeysField(ByteEnumField): |
| """ByteEnumField that picks valid values when fuzzed. """ |
| def randval(self): |
| return RandEnumKeys(self.i2s) |
| |
| |
| class ShortEnumKeysField(ShortEnumField): |
| """ShortEnumField that picks valid values when fuzzed. """ |
| def randval(self): |
| return RandEnumKeys(self.i2s) |
| |
| |
| class IntEnumKeysField(IntEnumField): |
| """IntEnumField that picks valid values when fuzzed. """ |
| def randval(self): |
| return RandEnumKeys(self.i2s) |
| |
| |
| # Little endian long field |
| class LELongField(Field): |
| def __init__(self, name, default): |
| Field.__init__(self, name, default, "<Q") |
| |
| # Little endian fixed length field |
| class LEFieldLenField(FieldLenField): |
| def __init__(self, name, default, length_of=None, fmt = "<H", count_of=None, adjust=lambda pkt,x:x, fld=None): |
| FieldLenField.__init__(self, name, default, length_of=length_of, fmt=fmt, count_of=count_of, fld=fld, adjust=adjust) |
| |
| |
| class FlagValue(object): |
| __slots__ = ["value", "names", "multi"] |
| def _fixvalue(self, value): |
| if isinstance(value, six.string_types): |
| value = value.split('+') if self.multi else list(value) |
| if isinstance(value, list): |
| y = 0 |
| for i in value: |
| y |= 1 << self.names.index(i) |
| value = y |
| return None if value is None else int(value) |
| def __init__(self, value, names): |
| self.multi = isinstance(names, list) |
| self.names = names |
| self.value = self._fixvalue(value) |
| def __hash__(self): |
| return hash(self.value) |
| def __int__(self): |
| return self.value |
| def __eq__(self, other): |
| return self.value == self._fixvalue(other) |
| def __lt__(self, other): |
| return self.value < self._fixvalue(other) |
| def __le__(self, other): |
| return self.value <= self._fixvalue(other) |
| def __gt__(self, other): |
| return self.value > self._fixvalue(other) |
| def __ge__(self, other): |
| return self.value >= self._fixvalue(other) |
| def __ne__(self, other): |
| return self.value != self._fixvalue(other) |
| def __and__(self, other): |
| return self.__class__(self.value & self._fixvalue(other), self.names) |
| __rand__ = __and__ |
| def __or__(self, other): |
| return self.__class__(self.value | self._fixvalue(other), self.names) |
| __ror__ = __or__ |
| def __lshift__(self, other): |
| return self.value << self._fixvalue(other) |
| def __rshift__(self, other): |
| return self.value >> self._fixvalue(other) |
| def __nonzero__(self): |
| return bool(self.value) |
| __bool__ = __nonzero__ |
| def flagrepr(self): |
| warning("obj.flagrepr() is obsolete. Use str(obj) instead.") |
| return str(self) |
| def __str__(self): |
| i = 0 |
| r = [] |
| x = int(self) |
| while x: |
| if x & 1: |
| r.append(self.names[i]) |
| i += 1 |
| x >>= 1 |
| return ("+" if self.multi else "").join(r) |
| def __repr__(self): |
| return "<Flag %d (%s)>" % (self, self) |
| def __deepcopy__(self, memo): |
| return self.__class__(int(self), self.names) |
| def __getattr__(self, attr): |
| if attr in self.__slots__: |
| return super(FlagValue, self).__getattr__(attr) |
| try: |
| if self.multi: |
| return bool((2 ** self.names.index(attr)) & int(self)) |
| return all(bool((2 ** self.names.index(flag)) & int(self)) |
| for flag in attr) |
| except ValueError: |
| return super(FlagValue, self).__getattr__(attr) |
| def __setattr__(self, attr, value): |
| if attr == "value" and not isinstance(value, six.integer_types): |
| raise ValueError(value) |
| if attr in self.__slots__: |
| return super(FlagValue, self).__setattr__(attr, value) |
| if attr in self.names: |
| if value: |
| self.value |= (2 ** self.names.index(attr)) |
| else: |
| self.value &= ~(2 ** self.names.index(attr)) |
| else: |
| return super(FlagValue, self).__setattr__(attr, value) |
| def copy(self): |
| return self.__class__(self.value, self.names) |
| |
| |
| class FlagsField(BitField): |
| """ Handle Flag type field |
| |
| Make sure all your flags have a label |
| |
| Example: |
| >>> from scapy.packet import Packet |
| >>> class FlagsTest(Packet): |
| fields_desc = [FlagsField("flags", 0, 8, ["f0", "f1", "f2", "f3", "f4", "f5", "f6", "f7"])] |
| >>> FlagsTest(flags=9).show2() |
| ###[ FlagsTest ]### |
| flags = f0+f3 |
| >>> FlagsTest(flags=0).show2().strip() |
| ###[ FlagsTest ]### |
| flags = |
| |
| :param name: field's name |
| :param default: default value for the field |
| :param size: number of bits in the field |
| :param names: (list or dict) label for each flag, Least Significant Bit tag's name is written first |
| """ |
| ismutable = True |
| __slots__ = ["multi", "names"] |
| |
| def __init__(self, name, default, size, names): |
| self.multi = isinstance(names, list) |
| self.names = names |
| BitField.__init__(self, name, default, size) |
| |
| def _fixup_val(self, x): |
| """Returns a FlagValue instance when needed. Internal method, to be |
| used in *2i() and i2*() methods. |
| |
| """ |
| if isinstance(x, (list, tuple)): |
| return type(x)( |
| v if v is None or isinstance(v, FlagValue) |
| else FlagValue(v, self.names) |
| for v in x |
| ) |
| return x if x is None or isinstance(x, FlagValue) else FlagValue(x, self.names) |
| |
| def any2i(self, pkt, x): |
| return self._fixup_val(super(FlagsField, self).any2i(pkt, x)) |
| |
| def m2i(self, pkt, x): |
| return self._fixup_val(super(FlagsField, self).m2i(pkt, x)) |
| |
| def i2h(self, pkt, x): |
| return self._fixup_val(super(FlagsField, self).i2h(pkt, x)) |
| |
| def i2repr(self, pkt, x): |
| if isinstance(x, (list, tuple)): |
| return repr(type(x)( |
| None if v is None else str(self._fixup_val(v)) for v in x |
| )) |
| return None if x is None else str(self._fixup_val(x)) |
| |
| |
| MultiFlagsEntry = collections.namedtuple('MultiFlagEntry', ['short', 'long']) |
| |
| |
| class MultiFlagsField(BitField): |
| __slots__ = FlagsField.__slots__ + ["depends_on"] |
| |
| def __init__(self, name, default, size, names, depends_on): |
| self.names = names |
| self.depends_on = depends_on |
| super(MultiFlagsField, self).__init__(name, default, size) |
| |
| def any2i(self, pkt, x): |
| assert isinstance(x, six.integer_types + (set,)), 'set expected' |
| |
| if pkt is not None: |
| if isinstance(x, six.integer_types): |
| x = self.m2i(pkt, x) |
| else: |
| v = self.depends_on(pkt) |
| if v is not None: |
| assert v in self.names, 'invalid dependency' |
| these_names = self.names[v] |
| s = set() |
| for i in x: |
| for val in six.itervalues(these_names): |
| if val.short == i: |
| s.add(i) |
| break |
| else: |
| assert False, 'Unknown flag "{}" with this dependency'.format(i) |
| continue |
| x = s |
| return x |
| |
| def i2m(self, pkt, x): |
| v = self.depends_on(pkt) |
| if v in self.names: |
| these_names = self.names[v] |
| else: |
| these_names = {} |
| |
| r = 0 |
| for flag_set in x: |
| for i, val in six.iteritems(these_names): |
| if val.short == flag_set: |
| r |= 1 << i |
| break |
| else: |
| r |= 1 << int(flag_set[len('bit '):]) |
| return r |
| |
| def m2i(self, pkt, x): |
| v = self.depends_on(pkt) |
| if v in self.names: |
| these_names = self.names[v] |
| else: |
| these_names = {} |
| |
| r = set() |
| i = 0 |
| |
| while x: |
| if x & 1: |
| if i in these_names: |
| r.add(these_names[i].short) |
| else: |
| r.add('bit {}'.format(i)) |
| x >>= 1 |
| i += 1 |
| return r |
| |
| def i2repr(self, pkt, x): |
| v = self.depends_on(pkt) |
| if v in self.names: |
| these_names = self.names[v] |
| else: |
| these_names = {} |
| |
| r = set() |
| for flag_set in x: |
| for i in six.itervalues(these_names): |
| if i.short == flag_set: |
| r.add("{} ({})".format(i.long, i.short)) |
| break |
| else: |
| r.add(flag_set) |
| return repr(r) |
| |
| |
| class FixedPointField(BitField): |
| __slots__ = ['frac_bits'] |
| def __init__(self, name, default, size, frac_bits=16): |
| self.frac_bits = frac_bits |
| BitField.__init__(self, name, default, size) |
| |
| def any2i(self, pkt, val): |
| if val is None: |
| return val |
| ival = int(val) |
| fract = int( (val-ival) * 2**self.frac_bits ) |
| return (ival << self.frac_bits) | fract |
| |
| def i2h(self, pkt, val): |
| int_part = val >> self.frac_bits |
| frac_part = val & (1 << self.frac_bits) - 1 |
| frac_part /= 2.0**self.frac_bits |
| return int_part+frac_part |
| def i2repr(self, pkt, val): |
| return self.i2h(pkt, val) |
| |
| |
| # Base class for IPv4 and IPv6 Prefixes inspired by IPField and IP6Field. |
| # Machine values are encoded in a multiple of wordbytes bytes. |
| class _IPPrefixFieldBase(Field): |
| __slots__ = ["wordbytes", "maxbytes", "aton", "ntoa", "length_from"] |
| def __init__(self, name, default, wordbytes, maxbytes, aton, ntoa, length_from): |
| self.wordbytes = wordbytes |
| self.maxbytes = maxbytes |
| self.aton = aton |
| self.ntoa = ntoa |
| Field.__init__(self, name, default, "%is" % self.maxbytes) |
| self.length_from = length_from |
| |
| def _numbytes(self, pfxlen): |
| wbits= self.wordbytes * 8 |
| return ((pfxlen + (wbits - 1)) // wbits) * self.wordbytes |
| |
| def h2i(self, pkt, x): |
| # "fc00:1::1/64" -> ("fc00:1::1", 64) |
| [pfx,pfxlen]= x.split('/') |
| self.aton(pfx) # check for validity |
| return (pfx, int(pfxlen)) |
| |
| |
| def i2h(self, pkt, x): |
| # ("fc00:1::1", 64) -> "fc00:1::1/64" |
| (pfx,pfxlen)= x |
| return "%s/%i" % (pfx,pfxlen) |
| |
| def i2m(self, pkt, x): |
| # ("fc00:1::1", 64) -> (b"\xfc\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01", 64) |
| (pfx,pfxlen)= x |
| s= self.aton(pfx); |
| return (s[:self._numbytes(pfxlen)], pfxlen) |
| |
| def m2i(self, pkt, x): |
| # (b"\xfc\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01", 64) -> ("fc00:1::1", 64) |
| (s,pfxlen)= x |
| |
| if len(s) < self.maxbytes: |
| s= s + (b"\0" * (self.maxbytes - len(s))) |
| return (self.ntoa(s), pfxlen) |
| |
| def any2i(self, pkt, x): |
| if x is None: |
| return (self.ntoa(b"\0"*self.maxbytes), 1) |
| |
| return self.h2i(pkt,x) |
| |
| def i2len(self, pkt, x): |
| (_,pfxlen)= x |
| return pfxlen |
| |
| def addfield(self, pkt, s, val): |
| (rawpfx,pfxlen)= self.i2m(pkt,val) |
| fmt= "!%is" % self._numbytes(pfxlen) |
| return s+struct.pack(fmt, rawpfx) |
| |
| def getfield(self, pkt, s): |
| pfxlen= self.length_from(pkt) |
| numbytes= self._numbytes(pfxlen) |
| fmt= "!%is" % numbytes |
| return s[numbytes:], self.m2i(pkt, (struct.unpack(fmt, s[:numbytes])[0], pfxlen)) |
| |
| |
| class IPPrefixField(_IPPrefixFieldBase): |
| def __init__(self, name, default, wordbytes=1, length_from= None): |
| _IPPrefixFieldBase.__init__(self, name, default, wordbytes, 4, inet_aton, inet_ntoa, length_from) |
| |
| |
| class IP6PrefixField(_IPPrefixFieldBase): |
| def __init__(self, name, default, wordbytes= 1, length_from= None): |
| _IPPrefixFieldBase.__init__(self, name, default, wordbytes, 16, lambda a: inet_pton(socket.AF_INET6, a), lambda n: inet_ntop(socket.AF_INET6, n), length_from) |
| |
| class UTCTimeField(IntField): |
| __slots__ = ["epoch", "delta", "strf", "use_nano"] |
| def __init__(self, name, default, epoch=None, use_nano=False, strf="%a, %d %b %Y %H:%M:%S +0000"): |
| IntField.__init__(self, name, default) |
| if epoch is None: |
| mk_epoch = EPOCH |
| else: |
| mk_epoch = time.mktime(epoch) |
| self.epoch = mk_epoch |
| self.delta = mk_epoch - EPOCH |
| self.strf = strf |
| self.use_nano = use_nano |
| def i2repr(self, pkt, x): |
| if x is None: |
| x = 0 |
| elif self.use_nano: |
| x = x/1e9 |
| x = int(x) + self.delta |
| t = time.strftime(self.strf, time.gmtime(x)) |
| return "%s (%d)" % (t, x) |
| def i2m(self, pkt, x): |
| return int(x) if x != None else 0 |
