Merge pull request secdev#631 from gpotter2/py3-set2-4

[Python 2 to 3] Fix overwritten imports (small)

scapy/autorun.py

@@ -7,7 +7,7 @@
 Run commands when the Scapy interpreter starts.
 """
 
-import code,sys
+import code, sys, importlib
 from scapy.config import conf
 from scapy.themes import *
 from scapy.error import Scapy_Exception
@@ -36,13 +36,16 @@ def showtraceback(self, *args, **kargs):
         return code.InteractiveInterpreter.showtraceback(self, *args, **kargs)
 
 
-def autorun_commands(cmds,my_globals=None,verb=0):
+def autorun_commands(cmds, my_globals=None, ignore_globals=None, verb=0):
     sv = conf.verb
     import __builtin__
     try:
         try:
             if my_globals is None:
-                my_globals = __import__("scapy.all").all.__dict__
+                my_globals = importlib.import_module(".all", "scapy").__dict__
+                if ignore_globals:
+                    for ig in ignore_globals:
+                        my_globals.pop(ig, None)
             conf.verb = verb
             interp = ScapyAutorunInterpreter(my_globals)
             cmd = ""

scapy/layers/all.py

@@ -7,15 +7,25 @@
 All layers. Configurable with conf.load_layers.
 """
 
+import __builtin__
 from scapy.config import conf
 from scapy.error import log_loading
-import logging
+import logging, importlib
+ignored = list(__builtin__.__dict__.keys()) + ["sys"]
 log = logging.getLogger("scapy.loading")
 
 __all__ = []
 
+
+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
+
 def _import_star(m):
-    mod = __import__(m, globals(), locals())
+    mod = importlib.import_module("." + m, "scapy.layers")
     if '__all__' in mod.__dict__:
         # only import the exported symbols in __all__
         for name in mod.__dict__['__all__']:
@@ -24,7 +34,7 @@ def _import_star(m):
     else:
         # import all the non-private symbols
         for name, sym in mod.__dict__.iteritems():
-            if name[0] != '_':
+            if _validate_local(name):
                 __all__.append(name)
                 globals()[name] = sym
 

scapy/main.py

@@ -11,8 +11,10 @@
 import glob
 import types
 import gzip
+import importlib
 import cPickle
 import __builtin__
+ignored = list(__builtin__.__dict__.keys())
 
 from scapy.error import *
 
@@ -35,6 +37,12 @@ def _read_config_file(cf):
     except Exception as e:
         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")
@@ -58,15 +66,15 @@ def _usage():
 
 def _load(module):
     try:
-        mod = __import__(module,globals(),locals(),".")
+        mod = importlib.import_module(module)
         if '__all__' in mod.__dict__:
             # import listed symbols
             for name in mod.__dict__['__all__']:
                 __builtin__.__dict__[name] = mod.__dict__[name]
         else:
             # only import non-private symbols
             for name, sym in mod.__dict__.iteritems():
-                if name[0] != '_':
+                if _validate_local(name):
                     __builtin__.__dict__[name] = sym
     except Exception as e:
         log_interactive.error(e)
@@ -79,7 +87,7 @@ def load_layer(name):
 
 def load_contrib(name):
     try:
-        __import__("scapy.contrib." + name)
+        importlib.import_module("scapy.contrib." + name)
         _load("scapy.contrib." + name)
     except ImportError:
         # if layer not found in contrib, try in layers
@@ -180,11 +188,11 @@ def init_session(session_name, mydict=None):
     global session
     global globkeys
 
-    scapy_builtins = __import__("all",globals(),locals(),".").__dict__
+    scapy_builtins = importlib.import_module(".all", "scapy").__dict__
     for name, sym in scapy_builtins.iteritems():
-        if name [0] != '_':
+        if _validate_local(name):
             __builtin__.__dict__[name] = sym
-    globkeys = scapy_builtins.keys()
+    globkeys = list(scapy_builtins.keys())
     globkeys.append("scapy_session")
     scapy_builtins=None # XXX replace with "with" statement
     if mydict is not None:

scapy/tools/UTscapy.py

@@ -7,7 +7,7 @@
 Unit testing infrastructure for Scapy
 """
 
-import sys, getopt, imp, glob
+import sys, getopt, imp, glob, importlib
 import bz2, base64, os.path, time, traceback, zlib, sha
 from scapy.consts import WINDOWS
 
@@ -355,17 +355,17 @@ def remove_empty_testsets(test_campaign):
 
 #### RUN CAMPAIGN #####
 
-def run_campaign(test_campaign, get_interactive_session, verb=3):
+def run_campaign(test_campaign, get_interactive_session, verb=3, ignore_globals=None):
     if WINDOWS:
         # Add a route to 127.0.0.1 and ::1
         from scapy.arch.windows import route_add_loopback
         route_add_loopback()
     passed=failed=0
     if test_campaign.preexec:
-        test_campaign.preexec_output = get_interactive_session(test_campaign.preexec.strip())[0]
+        test_campaign.preexec_output = get_interactive_session(test_campaign.preexec.strip(), ignore_globals=ignore_globals)[0]
     for testset in test_campaign:
         for t in testset:
-            t.output,res = get_interactive_session(t.test.strip())
+            t.output,res = get_interactive_session(t.test.strip(), ignore_globals=ignore_globals)
             the_res = False
             try:
                 if res is None or res:
@@ -607,7 +607,7 @@ def usage():
 #### MAIN ####
 
 def execute_campaign(TESTFILE, OUTPUTFILE, PREEXEC, NUM, KW_OK, KW_KO, DUMP,
-                     FORMAT, VERB, ONLYFAILED, CRC, autorun_func, pos_begin=0):
+                     FORMAT, VERB, ONLYFAILED, CRC, autorun_func, pos_begin=0, ignore_globals=None):
     # Parse test file
     test_campaign = parse_campaign_file(TESTFILE)
 
@@ -637,7 +637,7 @@ def execute_campaign(TESTFILE, OUTPUTFILE, PREEXEC, NUM, KW_OK, KW_KO, DUMP,
 
     # Run tests
     test_campaign.output_file = OUTPUTFILE
-    result = run_campaign(test_campaign, autorun_func[FORMAT], verb=VERB)
+    result = run_campaign(test_campaign, autorun_func[FORMAT], verb=VERB, ignore_globals=None)
 
     # Shrink passed
     if ONLYFAILED:
@@ -672,6 +672,7 @@ def resolve_testfiles(TESTFILES):
 
 def main(argv):
     import __builtin__
+    ignore_globals = list(__builtin__.__dict__.keys()) + ["sys"]
 
     # Parse arguments
 
@@ -815,7 +816,7 @@ def main(argv):
         output, result, campaign = execute_campaign(open(TESTFILE), OUTPUTFILE,
                                           PREEXEC, NUM, KW_OK, KW_KO,
                                           DUMP, FORMAT, VERB, ONLYFAILED,
-                                          CRC, autorun_func, pos_begin)
+                                          CRC, autorun_func, pos_begin, ignore_globals)
         runned_campaigns.append(campaign)
         pos_begin = campaign.end_pos
         if UNIQUE: