## 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
|
|
"""
|
Implementation of the configuration object.
|
"""
|
|
from __future__ import absolute_import
|
from __future__ import print_function
|
import os,time,socket,sys
|
|
from scapy import VERSION
|
from scapy.data import *
|
from scapy import base_classes
|
from scapy.themes import NoTheme, apply_ipython_style
|
from scapy.error import log_scapy
|
import scapy.modules.six as six
|
|
############
|
## Config ##
|
############
|
|
class ConfClass(object):
|
def configure(self, cnf):
|
self.__dict__ = cnf.__dict__.copy()
|
def __repr__(self):
|
return str(self)
|
def __str__(self):
|
s = ""
|
keys = self.__class__.__dict__.copy()
|
keys.update(self.__dict__)
|
keys = sorted(keys)
|
for i in keys:
|
if i[0] != "_":
|
r = repr(getattr(self, i))
|
r = " ".join(r.split())
|
wlen = 76-max(len(i),10)
|
if len(r) > wlen:
|
r = r[:wlen-3]+"..."
|
s += "%-10s = %s\n" % (i, r)
|
return s[:-1]
|
|
class Interceptor(object):
|
def __init__(self, name, default, hook, args=None, kargs=None):
|
self.name = name
|
self.intname = "_intercepted_%s" % name
|
self.default=default
|
self.hook = hook
|
self.args = args if args is not None else []
|
self.kargs = kargs if kargs is not None else {}
|
def __get__(self, obj, typ=None):
|
if not hasattr(obj, self.intname):
|
setattr(obj, self.intname, self.default)
|
return getattr(obj, self.intname)
|
def __set__(self, obj, val):
|
setattr(obj, self.intname, val)
|
self.hook(self.name, val, *self.args, **self.kargs)
|
|
|
class ProgPath(ConfClass):
|
pdfreader = "acroread"
|
psreader = "gv"
|
dot = "dot"
|
display = "display"
|
tcpdump = "tcpdump"
|
tcpreplay = "tcpreplay"
|
hexedit = "hexer"
|
tshark = "tshark"
|
wireshark = "wireshark"
|
ifconfig = "ifconfig"
|
|
|
class ConfigFieldList:
|
def __init__(self):
|
self.fields = set()
|
self.layers = set()
|
@staticmethod
|
def _is_field(f):
|
return hasattr(f, "owners")
|
def _recalc_layer_list(self):
|
self.layers = {owner for f in self.fields for owner in f.owners}
|
def add(self, *flds):
|
self.fields |= {f for f in flds if self._is_field(f)}
|
self._recalc_layer_list()
|
def remove(self, *flds):
|
self.fields -= set(flds)
|
self._recalc_layer_list()
|
def __contains__(self, elt):
|
if isinstance(elt, base_classes.Packet_metaclass):
|
return elt in self.layers
|
return elt in self.fields
|
def __repr__(self):
|
return "<%s [%s]>" % (self.__class__.__name__," ".join(str(x) for x in self.fields))
|
|
class Emphasize(ConfigFieldList):
|
pass
|
|
class Resolve(ConfigFieldList):
|
pass
|
|
|
class Num2Layer:
|
def __init__(self):
|
self.num2layer = {}
|
self.layer2num = {}
|
|
def register(self, num, layer):
|
self.register_num2layer(num, layer)
|
self.register_layer2num(num, layer)
|
|
def register_num2layer(self, num, layer):
|
self.num2layer[num] = layer
|
def register_layer2num(self, num, layer):
|
self.layer2num[layer] = num
|
|
def __getitem__(self, item):
|
if isinstance(item, base_classes.Packet_metaclass):
|
return self.layer2num[item]
|
return self.num2layer[item]
|
def __contains__(self, item):
|
if isinstance(item, base_classes.Packet_metaclass):
|
return item in self.layer2num
|
return item in self.num2layer
|
def get(self, item, default=None):
|
if item in self:
|
return self[item]
|
return default
|
|
def __repr__(self):
|
lst = []
|
for num,layer in six.iteritems(self.num2layer):
|
if layer in self.layer2num and self.layer2num[layer] == num:
|
dir = "<->"
|
else:
|
dir = " ->"
|
lst.append((num,"%#6x %s %-20s (%s)" % (num, dir, layer.__name__,
|
layer._name)))
|
for layer,num in six.iteritems(self.layer2num):
|
if num not in self.num2layer or self.num2layer[num] != layer:
|
lst.append((num,"%#6x <- %-20s (%s)" % (num, layer.__name__,
|
layer._name)))
|
lst.sort()
|
return "\n".join(y for x,y in lst)
|
|
|
class LayersList(list):
|
def __repr__(self):
|
s=[]
|
for l in self:
|
s.append("%-20s: %s" % (l.__name__,l.name))
|
return "\n".join(s)
|
def register(self, layer):
|
self.append(layer)
|
|
class CommandsList(list):
|
def __repr__(self):
|
s=[]
|
for l in sorted(self,key=lambda x:x.__name__):
|
if l.__doc__:
|
doc = l.__doc__.split("\n")[0]
|
else:
|
doc = "--"
|
s.append("%-20s: %s" % (l.__name__,doc))
|
return "\n".join(s)
|
def register(self, cmd):
|
self.append(cmd)
|
return cmd # return cmd so that method can be used as a decorator
|
|
def lsc():
|
print(repr(conf.commands))
|
|
class CacheInstance(dict, object):
|
__slots__ = ["timeout", "name", "_timetable", "__dict__"]
|
def __init__(self, name="noname", timeout=None):
|
self.timeout = timeout
|
self.name = name
|
self._timetable = {}
|
def flush(self):
|
self.__init__(name=self.name, timeout=self.timeout)
|
def __getitem__(self, item):
|
if item in self.__slots__:
|
return object.__getattribute__(self, item)
|
val = dict.__getitem__(self,item)
|
if self.timeout is not None:
|
t = self._timetable[item]
|
if time.time()-t > self.timeout:
|
raise KeyError(item)
|
return val
|
def get(self, item, default=None):
|
# overloading this method is needed to force the dict to go through
|
# the timetable check
|
try:
|
return self[item]
|
except KeyError:
|
return default
|
def __setitem__(self, item, v):
|
if item in self.__slots__:
|
return object.__setattr__(self, item, v)
|
self._timetable[item] = time.time()
|
dict.__setitem__(self, item,v)
|
def update(self, other):
|
for key, value in other.iteritems():
|
# We only update an element from `other` either if it does
|
# not exist in `self` or if the entry in `self` is older.
|
if key not in self or self._timetable[key] < other._timetable[key]:
|
dict.__setitem__(self, key, value)
|
self._timetable[key] = other._timetable[key]
|
def iteritems(self):
|
if self.timeout is None:
|
return six.iteritems(self.__dict__)
|
t0=time.time()
|
return ((k,v) for (k,v) in six.iteritems(self.__dict__) if t0-self._timetable[k] < self.timeout)
|
def iterkeys(self):
|
if self.timeout is None:
|
return six.iterkeys(self.__dict__)
|
t0=time.time()
|
return (k for k in six.iterkeys(self.__dict__) if t0-self._timetable[k] < self.timeout)
|
def __iter__(self):
|
return six.iterkeys(self.__dict__)
|
def itervalues(self):
|
if self.timeout is None:
|
return six.itervalues(self.__dict__)
|
t0=time.time()
|
return (v for (k,v) in six.iteritems(self.__dict__) if t0-self._timetable[k] < self.timeout)
|
def items(self):
|
if self.timeout is None:
|
return dict.items(self)
|
t0=time.time()
|
return [(k,v) for (k,v) in six.iteritems(self.__dict__) if t0-self._timetable[k] < self.timeout]
|
def keys(self):
|
if self.timeout is None:
|
return dict.keys(self)
|
t0=time.time()
|
return [k for k in six.iterkeys(self.__dict__) if t0-self._timetable[k] < self.timeout]
|
def values(self):
|
if self.timeout is None:
|
return six.values(self)
|
t0=time.time()
|
return [v for (k,v) in six.iteritems(self.__dict__) if t0-self._timetable[k] < self.timeout]
|
def __len__(self):
|
if self.timeout is None:
|
return dict.__len__(self)
|
return len(self.keys())
|
def summary(self):
|
return "%s: %i valid items. Timeout=%rs" % (self.name, len(self), self.timeout)
|
def __repr__(self):
|
s = []
|
if self:
|
mk = max(len(k) for k in six.iterkeys(self.__dict__))
|
fmt = "%%-%is %%s" % (mk+1)
|
for item in six.iteritems(self.__dict__):
|
s.append(fmt % item)
|
return "\n".join(s)
|
|
|
|
|
class NetCache:
|
def __init__(self):
|
self._caches_list = []
|
|
|
def add_cache(self, cache):
|
self._caches_list.append(cache)
|
setattr(self,cache.name,cache)
|
def new_cache(self, name, timeout=None):
|
c = CacheInstance(name=name, timeout=timeout)
|
self.add_cache(c)
|
def __delattr__(self, attr):
|
raise AttributeError("Cannot delete attributes")
|
def update(self, other):
|
for co in other._caches_list:
|
if hasattr(self, co.name):
|
getattr(self,co.name).update(co)
|
else:
|
self.add_cache(co.copy())
|
def flush(self):
|
for c in self._caches_list:
|
c.flush()
|
def __repr__(self):
|
return "\n".join(c.summary() for c in self._caches_list)
|
|
|
class LogLevel(object):
|
def __get__(self, obj, otype):
|
return obj._logLevel
|
def __set__(self,obj,val):
|
log_scapy.setLevel(val)
|
obj._logLevel = val
|
|
|
def isCryptographyValid():
|
"""
|
Check if the cryptography library is present, and if it is recent enough
|
for most usages in scapy (v1.7 or later).
|
"""
|
try:
|
import cryptography
|
except ImportError:
|
return False
|
from distutils.version import LooseVersion
|
return LooseVersion(cryptography.__version__) >= LooseVersion("1.7")
|
|
|
def isCryptographyAdvanced():
|
"""
|
Check if the cryptography library is present, and if it supports X25519,
|
ChaCha20Poly1305 and such (v2.0 or later).
|
"""
|
try:
|
import cryptography
|
except ImportError:
|
return False
|
from distutils.version import LooseVersion
|
lib_valid = LooseVersion(cryptography.__version__) >= LooseVersion("2.0")
|
if not lib_valid:
|
return False
|
|
try:
|
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
|
X25519PrivateKey.generate()
|
except:
|
return False
|
else:
|
return True
|
|
def isPyPy():
|
"""Returns either scapy is running under PyPy or not"""
|
try:
|
import __pypy__
|
return True
|
except ImportError:
|
return False
|
|
def _prompt_changer(attr, val):
|
"""Change the current prompt theme"""
|
try:
|
sys.ps1 = conf.color_theme.prompt(conf.prompt)
|
except:
|
pass
|
try:
|
apply_ipython_style(get_ipython())
|
except NameError:
|
pass
|
|
class Conf(ConfClass):
|
"""This object contains the configuration of Scapy.
|
session : filename where the session will be saved
|
interactive_shell : can be "ipython", "python" or "auto". Default: Auto
|
stealth : if 1, prevents any unwanted packet to go out (ARP, DNS, ...)
|
checkIPID: if 0, doesn't check that IPID matches between IP sent and ICMP IP citation received
|
if 1, checks that they either are equal or byte swapped equals (bug in some IP stacks)
|
if 2, strictly checks that they are equals
|
checkIPsrc: if 1, checks IP src in IP and ICMP IP citation match (bug in some NAT stacks)
|
checkIPinIP: if True, checks that IP-in-IP layers match. If False, do not
|
check IP layers that encapsulates another IP layer
|
check_TCPerror_seqack: if 1, also check that TCP seq and ack match the ones in ICMP citation
|
iff : selects the default output interface for srp() and sendp(). default:"eth0")
|
verb : level of verbosity, from 0 (almost mute) to 3 (verbose)
|
promisc : default mode for listening socket (to get answers if you spoof on a lan)
|
sniff_promisc : default mode for sniff()
|
filter : bpf filter added to every sniffing socket to exclude traffic from analysis
|
histfile : history file
|
padding : includes padding in disassembled packets
|
except_filter : BPF filter for packets to ignore
|
debug_match : when 1, store received packet that are not matched into debug.recv
|
route : holds the Scapy routing table and provides methods to manipulate it
|
warning_threshold : how much time between warnings from the same place
|
ASN1_default_codec: Codec used by default for ASN1 objects
|
mib : holds MIB direct access dictionary
|
resolve : holds list of fields for which resolution should be done
|
noenum : holds list of enum fields for which conversion to string should NOT be done
|
AS_resolver: choose the AS resolver class to use
|
extensions_paths: path or list of paths where extensions are to be looked for
|
contribs : a dict which can be used by contrib layers to store local configuration
|
debug_tls:When 1, print some TLS session secrets when they are computed.
|
"""
|
version = VERSION
|
session = ""
|
interactive = False
|
interactive_shell = ""
|
stealth = "not implemented"
|
iface = None
|
iface6 = None
|
layers = LayersList()
|
commands = CommandsList()
|
logLevel = LogLevel()
|
checkIPID = 0
|
checkIPsrc = 1
|
checkIPaddr = 1
|
checkIPinIP = True
|
check_TCPerror_seqack = 0
|
verb = 2
|
prompt = Interceptor("prompt", ">>> ", _prompt_changer)
|
promisc = 1
|
sniff_promisc = 1
|
raw_layer = None
|
raw_summary = False
|
default_l2 = None
|
l2types = Num2Layer()
|
l3types = Num2Layer()
|
L3socket = None
|
L2socket = None
|
L2listen = None
|
BTsocket = None
|
min_pkt_size = 60
|
histfile = os.getenv('SCAPY_HISTFILE',
|
os.path.join(os.path.expanduser("~"),
|
".scapy_history"))
|
padding = 1
|
except_filter = ""
|
debug_match = 0
|
debug_tls = 0
|
wepkey = ""
|
cache_iflist = {}
|
cache_ipaddrs = {}
|
route = None # Filed by route.py
|
route6 = None # Filed by route6.py
|
auto_fragment = 1
|
debug_dissector = 0
|
color_theme = Interceptor("color_theme", NoTheme(), _prompt_changer)
|
warning_threshold = 5
|
warning_next_only_once = False
|
prog = ProgPath()
|
resolve = Resolve()
|
noenum = Resolve()
|
emph = Emphasize()
|
use_pypy = isPyPy()
|
use_pcap = os.getenv("SCAPY_USE_PCAPDNET", "").lower().startswith("y")
|
use_dnet = os.getenv("SCAPY_USE_PCAPDNET", "").lower().startswith("y")
|
use_bpf = False
|
use_winpcapy = False
|
use_npcap = False
|
ipv6_enabled = socket.has_ipv6
|
ethertypes = ETHER_TYPES
|
protocols = IP_PROTOS
|
services_tcp = TCP_SERVICES
|
services_udp = UDP_SERVICES
|
extensions_paths = "."
|
manufdb = MANUFDB
|
stats_classic_protocols = []
|
stats_dot11_protocols = []
|
temp_files = []
|
netcache = NetCache()
|
geoip_city = '/usr/share/GeoIP/GeoIPCity.dat'
|
geoip_city_ipv6 = '/usr/share/GeoIP/GeoIPCityv6.dat'
|
load_layers = ["l2", "inet", "dhcp", "dns", "dot11", "gprs",
|
"hsrp", "inet6", "ir", "isakmp", "l2tp", "mgcp",
|
"mobileip", "netbios", "netflow", "ntp", "ppp", "pptp",
|
"radius", "rip", "rtp", "skinny", "smb", "snmp",
|
"tftp", "x509", "bluetooth", "dhcp6", "llmnr",
|
"sctp", "vrrp", "ipsec", "lltd", "vxlan", "eap"]
|
contribs = dict()
|
crypto_valid = isCryptographyValid()
|
crypto_valid_advanced = isCryptographyAdvanced()
|
fancy_prompt = True
|
|
|
if not Conf.ipv6_enabled:
|
log_scapy.warning("IPv6 support disabled in Python. Cannot load Scapy IPv6 layers.")
|
for m in ["inet6","dhcp6"]:
|
if m in Conf.load_layers:
|
Conf.load_layers.remove(m)
|
|
if not Conf.crypto_valid:
|
log_scapy.warning("Crypto-related methods disabled for IPsec, Dot11 "
|
"and TLS layers (needs python-cryptography v1.7+).")
|
|
conf=Conf()
|
conf.logLevel=30 # 30=Warning
|
|
|
def crypto_validator(func):
|
"""
|
This a decorator to be used for any method relying on the cryptography library.
|
Its behaviour depends on the 'crypto_valid' attribute of the global 'conf'.
|
"""
|
def func_in(*args, **kwargs):
|
if not conf.crypto_valid:
|
raise ImportError("Cannot execute crypto-related method! "
|
"Please install python-cryptography v1.7 or later.")
|
return func(*args, **kwargs)
|
return func_in
|
