Console & Config improvement

scapy/arch/pcapdnet.py

@@ -159,11 +159,10 @@ def load_winpcapy():
     except OSError:
         conf.use_winpcapy = False
         if conf.interactive:
-            log_loading.warning(conf.color_theme.format(
+            log_loading.critical(
                 "Npcap/Winpcap is not installed ! See "
-                "https://scapy.readthedocs.io/en/latest/installation.html#windows",  # noqa: E501
-                "black+bg_red"
-            ))
+                "https://scapy.readthedocs.io/en/latest/installation.html#windows"  # noqa: E501
+            )
 
     if conf.use_winpcapy:
         def get_if_list():

scapy/arch/windows/__init__.py

@@ -165,11 +165,18 @@ def _reload(self):
                                 env="SystemRoot")
         if self.wireshark:
             try:
-                manu_path = load_manuf(os.path.sep.join(self.wireshark.split(os.path.sep)[:-1]) + os.path.sep + "manuf")  # noqa: E501
+                new_manuf = load_manuf(
+                    os.path.sep.join(
+                        self.wireshark.split(os.path.sep)[:-1]
+                    ) + os.path.sep + "manuf"
+                )
             except (IOError, OSError):  # FileNotFoundError not available on Py2 - using OSError  # noqa: E501
                 log_loading.warning("Wireshark is installed, but cannot read manuf !")  # noqa: E501
-                manu_path = None
-            scapy.data.MANUFDB = conf.manufdb = manu_path
+                new_manuf = None
+            if new_manuf:
+                # Inject new ManufDB
+                conf.manufdb.__dict__.clear()
+                conf.manufdb.__dict__.update(new_manuf.__dict__)
 
 
 def _exec_cmd(command):

scapy/config.py

@@ -18,8 +18,6 @@
 
 from scapy import VERSION, base_classes
 from scapy.consts import DARWIN, WINDOWS, LINUX, BSD
-from scapy.data import ETHER_TYPES, IP_PROTOS, TCP_SERVICES, UDP_SERVICES, \
-    MANUFDB
 from scapy.error import log_scapy, warning, ScapyInvalidPlatformException
 from scapy.modules import six
 from scapy.themes import NoTheme, apply_ipython_style
@@ -53,7 +51,8 @@ def __str__(self):
 
 
 class Interceptor(object):
-    def __init__(self, name, default, hook, args=None, kargs=None):
+    def __init__(self, name=None, default=None,
+                 hook=None, args=None, kargs=None):
         self.name = name
         self.intname = "_intercepted_%s" % name
         self.default = default
@@ -76,6 +75,19 @@ def __set__(self, obj, val):
         self.hook(self.name, val, *self.args, **self.kargs)
 
 
+def _readonly(name):
+    default = Conf.__dict__[name].default
+    Interceptor.set_from_hook(conf, name, default)
+    raise ValueError("Read-only value !")
+
+
+ReadOnlyAttribute = functools.partial(
+    Interceptor,
+    hook=(lambda name, *args, **kwargs: _readonly(name))
+)
+ReadOnlyAttribute.__doc__ = "Read-only class attribute"
+
+
 class ProgPath(ConfClass):
     pdfreader = "open" if DARWIN else "xdg-open"
     psreader = "open" if DARWIN else "xdg-open"
@@ -211,6 +223,7 @@ def register(self, cmd):
 
 
 def lsc():
+    """Displays Scapy's default commands"""
     print(repr(conf.commands))
 
 
@@ -344,15 +357,6 @@ 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 _version_checker(module, minver):
     """Checks that module has a higher version that minver.
 
@@ -504,6 +508,11 @@ def _socket_changer(attr, val):
             raise
 
 
+def _loglevel_changer(attr, val):
+    """Handle a change of conf.logLevel"""
+    log_scapy.setLevel(val)
+
+
 class Conf(ConfClass):
     """This object contains the configuration of Scapy.
 session  : filename where the session will be saved
@@ -537,7 +546,7 @@ class Conf(ConfClass):
 debug_tls:When 1, print some TLS session secrets when they are computed.
 recv_poll_rate: how often to check for new packets. Defaults to 0.05s.
 """
-    version = VERSION
+    version = ReadOnlyAttribute("version", VERSION)
     session = ""
     interactive = False
     interactive_shell = ""
@@ -547,15 +556,15 @@ class Conf(ConfClass):
     layers = LayersList()
     commands = CommandsList()
     dot15d4_protocol = None  # Used in dot15d4.py
-    logLevel = LogLevel()
-    checkIPID = 0
-    checkIPsrc = 1
-    checkIPaddr = 1
+    logLevel = Interceptor("logLevel", log_scapy.level, _loglevel_changer)
+    checkIPID = False
+    checkIPsrc = True
+    checkIPaddr = True
     checkIPinIP = True
-    check_TCPerror_seqack = 0
+    check_TCPerror_seqack = False
     verb = 2
     prompt = Interceptor("prompt", ">>> ", _prompt_changer)
-    promisc = 1
+    promisc = True
     sniff_promisc = 1
     raw_layer = None
     raw_summary = False
@@ -574,21 +583,21 @@ class Conf(ConfClass):
                                       ".scapy_history"))
     padding = 1
     except_filter = ""
-    debug_match = 0
-    debug_tls = 0
+    debug_match = False
+    debug_tls = False
     wepkey = ""
     cache_iflist = {}
     route = None  # Filed by route.py
     route6 = None  # Filed by route6.py
-    auto_fragment = 1
-    debug_dissector = 0
+    auto_fragment = True
+    debug_dissector = False
     color_theme = Interceptor("color_theme", NoTheme(), _prompt_changer)
     warning_threshold = 5
     prog = ProgPath()
     resolve = Resolve()
     noenum = Resolve()
     emph = Emphasize()
-    use_pypy = isPyPy()
+    use_pypy = ReadOnlyAttribute("use_pypy", isPyPy())
     use_pcap = Interceptor(
         "use_pcap",
         os.getenv("SCAPY_USE_PCAPDNET", "").lower().startswith("y"),
@@ -600,12 +609,7 @@ class Conf(ConfClass):
     use_winpcapy = Interceptor("use_winpcapy", False, _socket_changer)
     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 = []
@@ -626,6 +630,25 @@ class Conf(ConfClass):
     auto_crop_tables = True
     recv_poll_rate = 0.05
 
+    def __getattr__(self, attr):
+        # Those are loaded on runtime to avoid import loops
+        if attr == "manufdb":
+            from scapy.data import MANUFDB
+            return MANUFDB
+        if attr == "ethertypes":
+            from scapy.data import ETHER_TYPES
+            return ETHER_TYPES
+        if attr == "protocols":
+            from scapy.data import IP_PROTOS
+            return IP_PROTOS
+        if attr == "services_udp":
+            from scapy.data import UDP_SERVICES
+            return UDP_SERVICES
+        if attr == "services_tcp":
+            from scapy.data import TCP_SERVICES
+            return TCP_SERVICES
+        return object.__getattr__(self, attr)
+
 
 if not Conf.ipv6_enabled:
     log_scapy.warning("IPv6 support disabled in Python. Cannot load Scapy IPv6 layers.")  # noqa: E501
@@ -634,7 +657,6 @@ class Conf(ConfClass):
             Conf.load_layers.remove(m)
 
 conf = Conf()
-conf.logLevel = 30  # 30=Warning
 
 
 def crypto_validator(func):

scapy/dadict.py

@@ -115,3 +115,8 @@ def iterkeys(self):
 
     def __len__(self):
         return len(self.__dict__)
+
+    def __nonzero__(self):
+        # Always has at least its name
+        return len(self.__dict__) > 1
+    __bool__ = __nonzero__

scapy/data.py

@@ -289,14 +289,11 @@ def load_manuf(filename):
 
 
 if WINDOWS:
-    ETHER_TYPES = load_ethertypes("ethertypes")
     IP_PROTOS = load_protocols(os.environ["SystemRoot"] + "\\system32\\drivers\\etc\\protocol")  # noqa: E501
     TCP_SERVICES, UDP_SERVICES = load_services(os.environ["SystemRoot"] + "\\system32\\drivers\\etc\\services")  # noqa: E501
-    # Default value, will be updated by arch.windows
-    try:
-        MANUFDB = load_manuf(os.environ["ProgramFiles"] + "\\wireshark\\manuf")
-    except (IOError, OSError):  # FileNotFoundError not available on Py2 - using OSError  # noqa: E501
-        MANUFDB = None
+    # Default values, will be updated by arch.windows
+    ETHER_TYPES = DADict()
+    MANUFDB = ManufDA()
 else:
     IP_PROTOS = load_protocols("/etc/protocols")
     ETHER_TYPES = load_ethertypes("/etc/ethertypes")

scapy/error.py

@@ -56,12 +56,37 @@ def filter(self, record):
         return 1
 
 
+# Inspired from python-colorbg (MIT)
+class ScapyColoredFormatter(logging.Formatter):
+    """A subclass of logging.Formatter that handles colors."""
+    levels_colored = {
+        'DEBUG': 'reset',
+        'INFO': 'reset',
+        'WARNING': 'bold+yellow',
+        'ERROR': 'bold+red',
+        'CRITICAL': 'bold+white+bg_red'
+    }
+
+    def format(self, record):
+        message = super(ScapyColoredFormatter, self).format(record)
+        from scapy.config import conf
+        message = conf.color_theme.format(
+            message,
+            self.levels_colored[record.levelname]
+        )
+        return message
+
+
 log_scapy = logging.getLogger("scapy")
+log_scapy.setLevel(logging.WARNING)
 log_scapy.addHandler(logging.NullHandler())
-log_runtime = logging.getLogger("scapy.runtime")          # logs at runtime
+# logs at runtime
+log_runtime = logging.getLogger("scapy.runtime")
 log_runtime.addFilter(ScapyFreqFilter())
-log_interactive = logging.getLogger("scapy.interactive")  # logs in interactive functions  # noqa: E501
-log_loading = logging.getLogger("scapy.loading")          # logs when loading Scapy  # noqa: E501
+# logs in interactive functions
+log_interactive = logging.getLogger("scapy.interactive")
+# logs when loading Scapy
+log_loading = logging.getLogger("scapy.loading")
 
 
 def warning(x, *args, **kargs):

scapy/extlib.py

@@ -54,7 +54,7 @@ def _test_pyx():
     if _test_pyx():
         PYX = 1
     else:
-        log_loading.warning("PyX dependencies are not installed ! Please install TexLive or MikTeX.")  # noqa: E501
+        log_loading.info("PyX dependencies are not installed ! Please install TexLive or MikTeX.")  # noqa: E501
         PYX = 0
 except ImportError:
     log_loading.info("Can't import PyX. Won't be able to use psdump() or pdfdump().")  # noqa: E501

scapy/layers/dhcp.py

@@ -346,8 +346,10 @@ class DHCP(Packet):
 @conf.commands.register
 def dhcp_request(iface=None, **kargs):
     """Send a DHCP discover request and return the answer"""
-    if conf.checkIPaddr != 0:
-        warning("conf.checkIPaddr is not 0, I may not be able to match the answer")  # noqa: E501
+    if conf.checkIPaddr:
+        warning(
+            "conf.checkIPaddr is enabled, may not be able to match the answer"
+        )
     if iface is None:
         iface = conf.iface
     fam, hw = get_if_raw_hwaddr(iface)