| ## 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 |
| |
| """ |
| Generators and packet meta classes. |
| """ |
| |
| ############### |
| ## Generators ## |
| ################ |
| |
| from __future__ import absolute_import |
| import re,random,socket |
| import types |
| from scapy.modules.six.moves import range |
| |
| class Gen(object): |
| __slots__ = [] |
| def __iter__(self): |
| return iter([]) |
| |
| class SetGen(Gen): |
| def __init__(self, values, _iterpacket=1): |
| self._iterpacket=_iterpacket |
| if isinstance(values, (list, BasePacketList)): |
| self.values = list(values) |
| elif (isinstance(values, tuple) and (2 <= len(values) <= 3) and \ |
| all(hasattr(i, "__int__") for i in values)): |
| # We use values[1] + 1 as stop value for (x)range to maintain |
| # the behavior of using tuples as field `values` |
| self.values = [range(*((int(values[0]), int(values[1]) + 1) |
| + tuple(int(v) for v in values[2:])))] |
| else: |
| self.values = [values] |
| def transf(self, element): |
| return element |
| def __iter__(self): |
| for i in self.values: |
| if (isinstance(i, Gen) and |
| (self._iterpacket or not isinstance(i,BasePacket))) or ( |
| isinstance(i, (range, types.GeneratorType))): |
| for j in i: |
| yield j |
| else: |
| yield i |
| def __repr__(self): |
| return "<SetGen %r>" % self.values |
| |
| class Net(Gen): |
| """Generate a list of IPs from a network address or a name""" |
| name = "ip" |
| ip_regex = re.compile(r"^(\*|[0-2]?[0-9]?[0-9](-[0-2]?[0-9]?[0-9])?)\.(\*|[0-2]?[0-9]?[0-9](-[0-2]?[0-9]?[0-9])?)\.(\*|[0-2]?[0-9]?[0-9](-[0-2]?[0-9]?[0-9])?)\.(\*|[0-2]?[0-9]?[0-9](-[0-2]?[0-9]?[0-9])?)(/[0-3]?[0-9])?$") |
| |
| @staticmethod |
| def _parse_digit(a,netmask): |
| netmask = min(8,max(netmask,0)) |
| if a == "*": |
| a = (0,256) |
| elif a.find("-") >= 0: |
| x, y = [int(d) for d in a.split('-')] |
| if x > y: |
| y = x |
| a = (x & (0xff<<netmask) , max(y, (x | (0xff>>(8-netmask))))+1) |
| else: |
| a = (int(a) & (0xff<<netmask),(int(a) | (0xff>>(8-netmask)))+1) |
| return a |
| |
| @classmethod |
| def _parse_net(cls, net): |
| tmp=net.split('/')+["32"] |
| if not cls.ip_regex.match(net): |
| tmp[0]=socket.gethostbyname(tmp[0]) |
| netmask = int(tmp[1]) |
| ret_list = [cls._parse_digit(x, y-netmask) for (x, y) in zip(tmp[0].split('.'), [8, 16, 24, 32])] |
| return ret_list, netmask |
| |
| def __init__(self, net): |
| self.repr=net |
| self.parsed,self.netmask = self._parse_net(net) |
| |
| def __str__(self): |
| try: |
| return next(self.__iter__()) |
| except StopIteration: |
| return None |
| |
| def __iter__(self): |
| for d in range(*self.parsed[3]): |
| for c in range(*self.parsed[2]): |
| for b in range(*self.parsed[1]): |
| for a in range(*self.parsed[0]): |
| yield "%i.%i.%i.%i" % (a,b,c,d) |
| def choice(self): |
| ip = [] |
| for v in self.parsed: |
| ip.append(str(random.randint(v[0],v[1]-1))) |
| return ".".join(ip) |
| |
| def __repr__(self): |
| return "Net(%r)" % self.repr |
| def __eq__(self, other): |
| if hasattr(other, "parsed"): |
| p2 = other.parsed |
| else: |
| p2,nm2 = self._parse_net(other) |
| return self.parsed == p2 |
| def __contains__(self, other): |
| if hasattr(other, "parsed"): |
| p2 = other.parsed |
| else: |
| p2,nm2 = self._parse_net(other) |
| for (a1,b1),(a2,b2) in zip(self.parsed,p2): |
| if a1 > a2 or b1 < b2: |
| return False |
| return True |
| def __rcontains__(self, other): |
| return self in self.__class__(other) |
| |
| |
| class OID(Gen): |
| name = "OID" |
| def __init__(self, oid): |
| self.oid = oid |
| self.cmpt = [] |
| fmt = [] |
| for i in oid.split("."): |
| if "-" in i: |
| fmt.append("%i") |
| self.cmpt.append(tuple(map(int, i.split("-")))) |
| else: |
| fmt.append(i) |
| self.fmt = ".".join(fmt) |
| def __repr__(self): |
| return "OID(%r)" % self.oid |
| def __iter__(self): |
| ii = [k[0] for k in self.cmpt] |
| while True: |
| yield self.fmt % tuple(ii) |
| i = 0 |
| while True: |
| if i >= len(ii): |
| raise StopIteration |
| if ii[i] < self.cmpt[i][1]: |
| ii[i]+=1 |
| break |
| else: |
| ii[i] = self.cmpt[i][0] |
| i += 1 |
| |
| |
| |
| ###################################### |
| ## Packet abstract and base classes ## |
| ###################################### |
| |
| class Packet_metaclass(type): |
| def __new__(cls, name, bases, dct): |
| if "fields_desc" in dct: # perform resolution of references to other packets |
| current_fld = dct["fields_desc"] |
| resolved_fld = [] |
| for f in current_fld: |
| if isinstance(f, Packet_metaclass): # reference to another fields_desc |
| for f2 in f.fields_desc: |
| resolved_fld.append(f2) |
| else: |
| resolved_fld.append(f) |
| else: # look for a fields_desc in parent classes |
| resolved_fld = None |
| for b in bases: |
| if hasattr(b,"fields_desc"): |
| resolved_fld = b.fields_desc |
| break |
| |
| if resolved_fld: # perform default value replacements |
| final_fld = [] |
| for f in resolved_fld: |
| if f.name in dct: |
| f = f.copy() |
| f.default = dct[f.name] |
| del(dct[f.name]) |
| final_fld.append(f) |
| |
| dct["fields_desc"] = final_fld |
| |
| if "__slots__" not in dct: |
| dct["__slots__"] = [] |
| for attr in ["name", "overload_fields"]: |
| try: |
| dct["_%s" % attr] = dct.pop(attr) |
| except KeyError: |
| pass |
| newcls = super(Packet_metaclass, cls).__new__(cls, name, bases, dct) |
| newcls.__all_slots__ = set( |
| attr |
| for cls in newcls.__mro__ if hasattr(cls, "__slots__") |
| for attr in cls.__slots__ |
| ) |
| |
| if hasattr(newcls, "aliastypes"): |
| newcls.aliastypes = [newcls] + newcls.aliastypes |
| else: |
| newcls.aliastypes = [newcls] |
| |
| if hasattr(newcls,"register_variant"): |
| newcls.register_variant() |
| for f in newcls.fields_desc: |
| if hasattr(f, "register_owner"): |
| f.register_owner(newcls) |
| from scapy import config |
| config.conf.layers.register(newcls) |
| return newcls |
| |
| def __getattr__(self, attr): |
| for k in self.fields_desc: |
| if k.name == attr: |
| return k |
| raise AttributeError(attr) |
| |
| def __call__(cls, *args, **kargs): |
| if "dispatch_hook" in cls.__dict__: |
| try: |
| cls = cls.dispatch_hook(*args, **kargs) |
| except: |
| from scapy import config |
| if config.conf.debug_dissector: |
| raise |
| cls = config.conf.raw_layer |
| i = cls.__new__(cls, cls.__name__, cls.__bases__, cls.__dict__) |
| i.__init__(*args, **kargs) |
| return i |
| |
| class Field_metaclass(type): |
| def __new__(cls, name, bases, dct): |
| if "__slots__" not in dct: |
| dct["__slots__"] = [] |
| newcls = super(Field_metaclass, cls).__new__(cls, name, bases, dct) |
| return newcls |
| |
| class NewDefaultValues(Packet_metaclass): |
| """NewDefaultValues is deprecated (not needed anymore) |
| |
| remove this: |
| __metaclass__ = NewDefaultValues |
| and it should still work. |
| """ |
| def __new__(cls, name, bases, dct): |
| from scapy.error import log_loading |
| import traceback |
| try: |
| for tb in traceback.extract_stack()+[("??",-1,None,"")]: |
| f,l,_,line = tb |
| if line.startswith("class"): |
| break |
| except: |
| f,l="??",-1 |
| raise |
| log_loading.warning("Deprecated (no more needed) use of NewDefaultValues (%s l. %i).", f, l) |
| |
| return super(NewDefaultValues, cls).__new__(cls, name, bases, dct) |
| |
| class BasePacket(Gen): |
| __slots__ = [] |
| |
| |
| ############################# |
| ## Packet list base class ## |
| ############################# |
| |
| class BasePacketList(object): |
| __slots__ = [] |
