## 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
|
|
"""
|
Main module for interactive startup.
|
"""
|
|
from __future__ import absolute_import
|
from __future__ import print_function
|
|
import sys, os, getopt, re, code
|
import gzip, glob
|
import importlib
|
import logging
|
from random import choice
|
import types
|
import io
|
|
# Never add any global import, in main.py, that would trigger a warning messsage
|
# before the console handlers gets added in interact()
|
from scapy.error import log_interactive, log_loading, log_scapy, warning
|
import scapy.modules.six as six
|
from scapy.themes import DefaultTheme, apply_ipython_style
|
|
IGNORED = list(six.moves.builtins.__dict__)
|
|
GLOBKEYS = []
|
|
LAYER_ALIASES = {
|
"tls": "tls.all"
|
}
|
|
QUOTES = [
|
("Craft packets like it is your last day on earth.", "Lao-Tze"),
|
("Craft packets like I craft my beer.", "Jean De Clerck"),
|
("Craft packets before they craft you.", "Socrate"),
|
("Craft me if you can.", "IPv6 layer"),
|
("To craft a packet, you have to be a packet, and learn how to swim in the "
|
"wires and in the waves.", "Jean-Claude Van Damme"),
|
]
|
|
def _probe_config_file(cf):
|
cf_path = os.path.join(os.path.expanduser("~"), cf)
|
try:
|
os.stat(cf_path)
|
except OSError:
|
return None
|
else:
|
return cf_path
|
|
def _read_config_file(cf, _globals=globals(), _locals=locals(), interactive=True):
|
"""Read a config file: execute a python file while loading scapy, that may contain
|
some pre-configured values.
|
|
If _globals or _locals are specified, they will be updated with the loaded vars.
|
This allows an external program to use the function. Otherwise, vars are only available
|
from inside the scapy console.
|
|
params:
|
- _globals: the globals() vars
|
- _locals: the locals() vars
|
- interactive: specified whether or not errors should be printed using the scapy console or
|
raised.
|
|
ex, content of a config.py file:
|
'conf.verb = 42\n'
|
Manual loading:
|
>>> _read_config_file("./config.py"))
|
>>> conf.verb
|
42
|
"""
|
log_loading.debug("Loading config file [%s]", cf)
|
try:
|
exec(compile(open(cf).read(), cf, 'exec'), _globals, _locals)
|
except IOError as e:
|
if interactive:
|
raise
|
log_loading.warning("Cannot read config file [%s] [%s]", cf, e)
|
except Exception as e:
|
if interactive:
|
raise
|
log_loading.exception("Error during evaluation of config file [%s]", cf)
|
|
def _validate_local(x):
|
"""Returns whether or not a variable should be imported.
|
Will return False for any default modules (sys), or if
|
they are detected as private vars (starting with a _)"""
|
global IGNORED
|
return x[0] != "_" and not x in IGNORED
|
|
DEFAULT_PRESTART_FILE = _probe_config_file(".scapy_prestart.py")
|
DEFAULT_STARTUP_FILE = _probe_config_file(".scapy_startup.py")
|
SESSION = None
|
|
def _usage():
|
print("""Usage: scapy.py [-s sessionfile] [-c new_startup_file] [-p new_prestart_file] [-C] [-P]
|
-C: do not read startup file
|
-P: do not read pre-startup file""")
|
sys.exit(0)
|
|
|
######################
|
## Extension system ##
|
######################
|
|
|
def _load(module, globals_dict=None, symb_list=None):
|
"""Loads a Python module to make variables, objects and functions
|
available globally.
|
|
The idea is to load the module using importlib, then copy the
|
symbols to the global symbol table.
|
|
"""
|
if globals_dict is None:
|
globals_dict = six.moves.builtins.__dict__
|
try:
|
mod = importlib.import_module(module)
|
if '__all__' in mod.__dict__:
|
# import listed symbols
|
for name in mod.__dict__['__all__']:
|
if symb_list is not None:
|
symb_list.append(name)
|
globals_dict[name] = mod.__dict__[name]
|
else:
|
# only import non-private symbols
|
for name, sym in six.iteritems(mod.__dict__):
|
if _validate_local(name):
|
if symb_list is not None:
|
symb_list.append(name)
|
globals_dict[name] = sym
|
except Exception:
|
log_interactive.error("Loading module %s", module, exc_info=True)
|
|
def load_module(name):
|
"""Loads a Scapy module to make variables, objects and functions
|
available globally.
|
|
"""
|
_load("scapy.modules."+name)
|
|
def load_layer(name, globals_dict=None, symb_list=None):
|
"""Loads a Scapy layer module to make variables, objects and functions
|
available globally.
|
|
"""
|
_load("scapy.layers." + LAYER_ALIASES.get(name, name),
|
globals_dict=globals_dict, symb_list=symb_list)
|
|
def load_contrib(name):
|
"""Loads a Scapy contrib module to make variables, objects and
|
functions available globally.
|
|
If no contrib module can be found with the given name, try to find
|
a layer module, since a contrib module may become a layer module.
|
|
"""
|
try:
|
importlib.import_module("scapy.contrib." + name)
|
_load("scapy.contrib." + name)
|
except ImportError:
|
# if layer not found in contrib, try in layers
|
load_layer(name)
|
|
def list_contrib(name=None):
|
if name is None:
|
name="*.py"
|
elif "*" not in name and "?" not in name and not name.endswith(".py"):
|
name += ".py"
|
name = os.path.join(os.path.dirname(__file__), "contrib", name)
|
for f in sorted(glob.glob(name)):
|
mod = os.path.basename(f)
|
if mod.startswith("__"):
|
continue
|
if mod.endswith(".py"):
|
mod = mod[:-3]
|
desc = { "description":"-", "status":"?", "name":mod }
|
for l in io.open(f, errors="replace"):
|
p = l.find("scapy.contrib.")
|
if p >= 0:
|
p += 14
|
q = l.find("=", p)
|
key = l[p:q].strip()
|
value = l[q+1:].strip()
|
desc[key] = value
|
print("%(name)-20s: %(description)-40s status=%(status)s" % desc)
|
|
|
|
|
|
|
##############################
|
## Session saving/restoring ##
|
##############################
|
|
def update_ipython_session(session):
|
"""Updates IPython session with a custom one"""
|
try:
|
get_ipython().user_ns.update(session)
|
except:
|
pass
|
|
def save_session(fname=None, session=None, pickleProto=-1):
|
"""Save current Scapy session to the file specified in the fname arg.
|
|
params:
|
- fname: file to save the scapy session in
|
- session: scapy session to use. If None, the console one will be used
|
- pickleProto: pickle proto version (default: -1 = latest)"""
|
from scapy import utils
|
if fname is None:
|
fname = conf.session
|
if not fname:
|
conf.session = fname = utils.get_temp_file(keep=True)
|
log_interactive.info("Use [%s] as session file" % fname)
|
|
if session is None:
|
try:
|
session = get_ipython().user_ns
|
except:
|
session = six.moves.builtins.__dict__["scapy_session"]
|
|
to_be_saved = session.copy()
|
if "__builtins__" in to_be_saved:
|
del(to_be_saved["__builtins__"])
|
|
for k in list(to_be_saved):
|
i = to_be_saved[k]
|
if hasattr(i, "__module__") and (k[0] == "_" or i.__module__.startswith("IPython")):
|
del(to_be_saved[k])
|
if isinstance(i, ConfClass):
|
del(to_be_saved[k])
|
elif isinstance(i, (type, type, types.ModuleType)):
|
if k[0] != "_":
|
log_interactive.error("[%s] (%s) can't be saved.", k, type(to_be_saved[k]))
|
del(to_be_saved[k])
|
|
try:
|
os.rename(fname, fname+".bak")
|
except OSError:
|
pass
|
|
f=gzip.open(fname,"wb")
|
six.moves.cPickle.dump(to_be_saved, f, pickleProto)
|
f.close()
|
del f
|
|
def load_session(fname=None):
|
"""Load current Scapy session from the file specified in the fname arg.
|
This will erase any existing session.
|
|
params:
|
- fname: file to load the scapy session from"""
|
if fname is None:
|
fname = conf.session
|
try:
|
s = six.moves.cPickle.load(gzip.open(fname,"rb"))
|
except IOError:
|
try:
|
s = six.moves.cPickle.load(open(fname,"rb"))
|
except IOError:
|
# Raise "No such file exception"
|
raise
|
|
scapy_session = six.moves.builtins.__dict__["scapy_session"]
|
scapy_session.clear()
|
scapy_session.update(s)
|
update_ipython_session(scapy_session)
|
|
log_loading.info("Loaded session [%s]" % fname)
|
|
def update_session(fname=None):
|
"""Update current Scapy session from the file specified in the fname arg.
|
|
params:
|
- fname: file to load the scapy session from"""
|
if fname is None:
|
fname = conf.session
|
try:
|
s = six.moves.cPickle.load(gzip.open(fname,"rb"))
|
except IOError:
|
s = six.moves.cPickle.load(open(fname,"rb"))
|
scapy_session = six.moves.builtins.__dict__["scapy_session"]
|
scapy_session.update(s)
|
update_ipython_session(scapy_session)
|
|
def init_session(session_name, mydict=None):
|
global SESSION
|
global GLOBKEYS
|
|
scapy_builtins = {k: v for k, v in six.iteritems(importlib.import_module(".all", "scapy").__dict__) if _validate_local(k)}
|
six.moves.builtins.__dict__.update(scapy_builtins)
|
GLOBKEYS.extend(scapy_builtins)
|
GLOBKEYS.append("scapy_session")
|
scapy_builtins=None # XXX replace with "with" statement
|
|
if session_name:
|
try:
|
os.stat(session_name)
|
except OSError:
|
log_loading.info("New session [%s]" % session_name)
|
else:
|
try:
|
try:
|
SESSION = six.moves.cPickle.load(gzip.open(session_name,"rb"))
|
except IOError:
|
SESSION = six.moves.cPickle.load(open(session_name,"rb"))
|
log_loading.info("Using session [%s]" % session_name)
|
except EOFError:
|
log_loading.error("Error opening session [%s]" % session_name)
|
except AttributeError:
|
log_loading.error("Error opening session [%s]. Attribute missing" % session_name)
|
|
if SESSION:
|
if "conf" in SESSION:
|
conf.configure(SESSION["conf"])
|
SESSION["conf"] = conf
|
else:
|
conf.session = session_name
|
SESSION = {"conf":conf}
|
else:
|
SESSION = {"conf": conf}
|
|
six.moves.builtins.__dict__["scapy_session"] = SESSION
|
|
if mydict is not None:
|
six.moves.builtins.__dict__["scapy_session"].update(mydict)
|
update_ipython_session(mydict)
|
GLOBKEYS.extend(mydict)
|
|
################
|
##### Main #####
|
################
|
|
def scapy_delete_temp_files():
|
for f in conf.temp_files:
|
try:
|
os.unlink(f)
|
except:
|
pass
|
del(conf.temp_files[:])
|
|
def _prepare_quote(quote, author, max_len=78):
|
"""This function processes a quote and returns a string that is ready
|
to be used in the fancy prompt.
|
|
"""
|
quote = quote.split(' ')
|
max_len -= 6
|
lines = []
|
cur_line = []
|
def _len(line):
|
return sum(len(elt) for elt in line) + len(line) - 1
|
while quote:
|
if not cur_line or (_len(cur_line) + len(quote[0]) - 1 <= max_len):
|
cur_line.append(quote.pop(0))
|
continue
|
lines.append(' | %s' % ' '.join(cur_line))
|
cur_line = []
|
if cur_line:
|
lines.append(' | %s' % ' '.join(cur_line))
|
cur_line = []
|
lines.append(' | %s-- %s' % (" " * (max_len - len(author) - 5), author))
|
return lines
|
|
def interact(mydict=None,argv=None,mybanner=None,loglevel=20):
|
global SESSION
|
global GLOBKEYS
|
|
console_handler = logging.StreamHandler()
|
console_handler.setFormatter(logging.Formatter("%(levelname)s: %(message)s"))
|
log_scapy.addHandler(console_handler)
|
|
from scapy.config import conf
|
conf.color_theme = DefaultTheme()
|
conf.interactive = True
|
if loglevel is not None:
|
conf.logLevel = loglevel
|
|
STARTUP_FILE = DEFAULT_STARTUP_FILE
|
PRESTART_FILE = DEFAULT_PRESTART_FILE
|
|
session_name = None
|
|
if argv is None:
|
argv = sys.argv
|
|
try:
|
opts = getopt.getopt(argv[1:], "hs:Cc:Pp:d")
|
for opt, parm in opts[0]:
|
if opt == "-h":
|
_usage()
|
elif opt == "-s":
|
session_name = parm
|
elif opt == "-c":
|
STARTUP_FILE = parm
|
elif opt == "-C":
|
STARTUP_FILE = None
|
elif opt == "-p":
|
PRESTART_FILE = parm
|
elif opt == "-P":
|
PRESTART_FILE = None
|
elif opt == "-d":
|
conf.logLevel = max(1, conf.logLevel-10)
|
|
if len(opts[1]) > 0:
|
raise getopt.GetoptError("Too many parameters : [%s]" % " ".join(opts[1]))
|
|
|
except getopt.GetoptError as msg:
|
log_loading.error(msg)
|
sys.exit(1)
|
|
init_session(session_name, mydict)
|
|
if STARTUP_FILE:
|
_read_config_file(STARTUP_FILE, interactive=True)
|
if PRESTART_FILE:
|
_read_config_file(PRESTART_FILE, interactive=True)
|
|
if conf.fancy_prompt:
|
|
the_logo = [
|
" ",
|
" aSPY//YASa ",
|
" apyyyyCY//////////YCa ",
|
" sY//////YSpcs scpCY//Pp ",
|
" ayp ayyyyyyySCP//Pp syY//C ",
|
" AYAsAYYYYYYYY///Ps cY//S",
|
" pCCCCY//p cSSps y//Y",
|
" SPPPP///a pP///AC//Y",
|
" A//A cyP////C",
|
" p///Ac sC///a",
|
" P////YCpc A//A",
|
" scccccp///pSP///p p//Y",
|
" sY/////////y caa S//P",
|
" cayCyayP//Ya pY/Ya",
|
" sY/PsY////YCc aC//Yp ",
|
" sc sccaCY//PCypaapyCP//YSs ",
|
" spCPY//////YPSps ",
|
" ccaacs ",
|
" ",
|
]
|
|
the_banner = [
|
"",
|
"",
|
" |",
|
" | Welcome to Scapy",
|
" | Version %s" % conf.version,
|
" |",
|
" | https://github.com/secdev/scapy",
|
" |",
|
" | Have fun!",
|
" |",
|
]
|
|
quote, author = choice(QUOTES)
|
the_banner.extend(_prepare_quote(quote, author, max_len=39))
|
the_banner.append(" |")
|
the_banner = "\n".join(
|
logo + banner for logo, banner in six.moves.zip_longest(
|
(conf.color_theme.logo(line) for line in the_logo),
|
(conf.color_theme.success(line) for line in the_banner),
|
fillvalue=""
|
)
|
)
|
else:
|
the_banner = "Welcome to Scapy (%s)" % conf.version
|
if mybanner is not None:
|
the_banner += "\n"
|
the_banner += mybanner
|
|
if not conf.interactive_shell or conf.interactive_shell.lower() in [
|
"ipython", "auto"
|
]:
|
try:
|
import IPython
|
from IPython.terminal.embed import InteractiveShellEmbed
|
except ImportError:
|
log_loading.warning(
|
"IPython not available. Using standard Python shell "
|
"instead.\nAutoCompletion, History are disabled."
|
)
|
IPYTHON = False
|
else:
|
IPYTHON = True
|
else:
|
IPYTHON = False
|
|
init_session(session_name, mydict)
|
|
if IPYTHON:
|
banner = the_banner + " using IPython %s\n" % IPython.__version__
|
try:
|
from traitlets.config.loader import Config
|
except ImportError:
|
log_loading.warning(
|
"traitlets not available. Some Scapy shell features won't be "
|
"available."
|
)
|
try:
|
ipshell = InteractiveShellEmbed(
|
banner1=banner,
|
user_ns=SESSION,
|
)
|
except:
|
code.interact(banner = the_banner, local=SESSION)
|
else:
|
cfg = Config()
|
try:
|
get_ipython
|
except NameError:
|
# Set "classic" prompt style when launched from run_scapy(.bat) files
|
# Register and apply scapy color+prompt style
|
apply_ipython_style(shell=cfg.TerminalInteractiveShell)
|
cfg.TerminalInteractiveShell.confirm_exit = False
|
cfg.TerminalInteractiveShell.separate_in = u''
|
cfg.TerminalInteractiveShell.hist_file = conf.histfile
|
# configuration can thus be specified here.
|
try:
|
ipshell = InteractiveShellEmbed(config=cfg,
|
banner1=banner,
|
hist_file=conf.histfile if conf.histfile else None,
|
user_ns=SESSION)
|
except (AttributeError, TypeError):
|
log_loading.warning("IPython too old. Won't support history and color style.")
|
try:
|
ipshell = InteractiveShellEmbed(
|
banner1=banner,
|
user_ns=SESSION,
|
)
|
except:
|
code.interact(banner = the_banner, local=SESSION)
|
ipshell(local_ns=SESSION)
|
else:
|
code.interact(banner = the_banner, local=SESSION)
|
|
if conf.session:
|
save_session(conf.session, SESSION)
|
|
for k in GLOBKEYS:
|
try:
|
del(six.moves.builtins.__dict__[k])
|
except:
|
pass
|
|
if __name__ == "__main__":
|
interact()
|
