[Hinty] Type hinting: core - Windows

.mypy/mypy_check.py

@@ -0,0 +1,32 @@
+# This file is part of Scapy
+# See http://www.secdev.org/projects/scapy for more information
+# Copyright (C) Philippe Biondi <phil@secdev.org>
+# Copyright (C) Gabriel Potter <gabriel@potter.fr>
+# This program is published under a GPLv2 license
+
+"""
+Performs Static typing checks over Scapy's codebase
+"""
+
+import io
+import os
+import sys
+
+from mypy.main import main as mypy_main
+
+# Load files
+
+with io.open("./.mypy/mypy_enabled.txt") as fd:
+    FILES = [l.strip() for l in fd.readlines() if l.strip() and l[0] != "#"]
+
+if not FILES:
+    print("No files specified. Arborting")
+    sys.exit(0)
+
+# Generate Mypy arguments
+
+ARGS = ["--py2"] + [os.path.abspath(f) for f in FILES]
+
+# Run Mypy over the files
+
+mypy_main(None, sys.stdout, sys.stderr, ARGS)

.mypy/mypy_enabled.txt

@@ -0,0 +1,47 @@
+# This file registers all files that have already been processed as part of
+# https://github.com/secdev/scapy/issues/2158, and therefore will be enforced
+# with unit tests in future development.
+
+# Style cheet: https://mypy.readthedocs.io/en/latest/cheat_sheet.html
+
+scapy/__init__.py
+scapy/ansmachine.py
+scapy/as_resolvers.py
+scapy/asn1/asn1.py
+scapy/asn1/ber.py
+scapy/asn1/mib.py
+scapy/asn1fields.py
+scapy/asn1packet.py
+scapy/automaton.py
+scapy/autorun.py
+scapy/base_classes.py
+scapy/compat.py
+scapy/config.py
+scapy/dadict.py
+scapy/data.py
+scapy/error.py
+scapy/extlib.py
+scapy/fields.py
+scapy/main.py
+scapy/modules/nmap.py
+scapy/modules/p0f.py
+scapy/packet.py
+scapy/pipetool.py
+scapy/plist.py
+scapy/pton_ntop.py
+scapy/route.py
+scapy/route6.py
+scapy/scapypipes.py
+scapy/sendrecv.py
+scapy/sessions.py
+scapy/supersocket.py
+scapy/themes.py
+scapy/utils.py
+scapy/utils6.py
+scapy/volatile.py
+
+scapy/arch/__init__.py
+scapy/arch/common.py
+scapy/arch/pcapdnet.py
+scapy/arch/windows/__init__.py
+scapy/arch/windows/structures.py

.travis.yml

@@ -6,7 +6,7 @@ matrix:
         - os: linux
           python: 3.6
           env:
-            - TOXENV=flake8,twine,docs,spell
+            - TOXENV=mypy,flake8,twine,docs,spell
 
         # Run as a regular user
         - os: linux

scapy/__init__.py

@@ -19,6 +19,7 @@
 
 
 def _version_from_git_describe():
+    # type: () -> str
     """
     Read the version from ``git describe``. It returns the latest tag with an
     optional suffix if the current directory is not exactly on the tag.
@@ -38,6 +39,9 @@ def _version_from_git_describe():
 
         >>> _version_from_git_describe()
         '2.3.2.dev346'
+
+    :raises CalledProcessError: if git is unavailable
+    :return: the scapy latest tag
     """
     if not os.path.isdir(os.path.join(os.path.dirname(_SCAPY_PKG_DIR), '.git')):  # noqa: E501
         raise ValueError('not in scapy git repo')
@@ -62,6 +66,11 @@ def _version_from_git_describe():
 
 
 def _version():
+    # () -> str
+    """Returns the Scapy version from multiple methods
+
+    :return: the scapy version
+    """
     version_file = os.path.join(_SCAPY_PKG_DIR, 'VERSION')
     try:
         tag = _version_from_git_describe()

scapy/ansmachine.py

@@ -14,9 +14,14 @@
 from __future__ import absolute_import
 from __future__ import print_function
 from scapy.sendrecv import send, sniff
+from scapy.packet import Packet
 from scapy.config import conf
 from scapy.error import log_interactive
 import scapy.modules.six as six
+from typing import Any
+from typing import Dict
+from typing import Union
+from typing import Tuple
 
 
 class ReferenceAM(type):
@@ -37,6 +42,7 @@ class AnsweringMachine(six.with_metaclass(ReferenceAM, object)):
     send_function = staticmethod(send)
 
     def __init__(self, **kargs):
+        # type: (**Any) -> None
         self.mode = 0
         if self.filter:
             kargs.setdefault("filter", self.filter)
@@ -52,12 +58,14 @@ def __init__(self, **kargs):
         self.optsend, self.optsniff = [{}, {}]
 
     def __getattr__(self, attr):
+        # type: (str) -> str
         for dct in [self.optam2, self.optam1]:
             if attr in dct:
                 return dct[attr]
         raise AttributeError(attr)
 
     def __setattr__(self, attr, val):
+        # type: (str, Union[Dict[str, int], int]) -> None
         mode = self.__dict__.get("mode", 0)
         if mode == 0:
             self.__dict__[attr] = val
@@ -68,6 +76,7 @@ def parse_options(self):
         pass
 
     def parse_all_options(self, mode, kargs):
+        # type: (int, Dict[str, Any]) -> Tuple[Dict, Dict[str, Any]]
         sniffopt = {}
         sendopt = {}
         for k in list(kargs):  # use list(): kargs is modified in the loop
@@ -101,9 +110,11 @@ def send_reply(self, reply):
         self.send_function(reply, **self.optsend)
 
     def print_reply(self, req, reply):
+        # type: (Packet, Packet) -> None
         print("%s ==> %s" % (req.summary(), reply.summary()))
 
     def reply(self, pkt):
+        # type: (Packet) -> None
         if not self.is_request(pkt):
             return
         reply = self.make_reply(pkt)
@@ -116,6 +127,7 @@ def run(self, *args, **kargs):
         self(*args, **kargs)
 
     def __call__(self, *args, **kargs):
+        # type: (*Any, **Any) -> None
         optsend, optsniff = self.parse_all_options(2, kargs)
         self.optsend = self.defoptsend.copy()
         self.optsend.update(optsend)
@@ -128,4 +140,5 @@ def __call__(self, *args, **kargs):
             print("Interrupted by user")
 
     def sniff(self):
+        # type: () -> None
         sniff(**self.optsniff)

scapy/arch/__init__.py

@@ -29,10 +29,12 @@ def str2mac(s):
 
 
 def get_if_addr(iff):
+    # type: (Any) -> str
     return inet_ntop(socket.AF_INET, get_if_raw_addr(iff))
 
 
 def get_if_hwaddr(iff):
+    # type: (Any) -> str
     addrfamily, mac = get_if_raw_hwaddr(iff)  # noqa: F405
     if addrfamily in [ARPHDR_ETHER, ARPHDR_LOOPBACK]:
         return str2mac(mac)
@@ -74,6 +76,7 @@ def get_if_hwaddr(iff):
 
 
 def get_if_addr6(iff):
+    # type: (Any) -> str
     """
     Returns the main global unicast address associated with provided
     interface, in human readable form. If no global address is found,
@@ -84,6 +87,7 @@ def get_if_addr6(iff):
 
 
 def get_if_raw_addr6(iff):
+    # type: (Any) -> bytes
     """
     Returns the main global unicast address associated with provided
     interface, in network format. If no global address is found, None

scapy/arch/common.py

@@ -22,6 +22,13 @@
 from scapy.consts import OPENBSD
 import scapy.modules.six as six
 
+from typing import Any
+from typing import Callable
+from typing import Dict
+from typing import Optional
+from typing import Tuple
+from typing import Union
+
 if not WINDOWS:
     from fcntl import ioctl
 
@@ -63,13 +70,17 @@ def get_if(iff, cmd):
 
 # SOCKET UTILS
 
-def _select_nonblock(sockets, remain=None):
+def _select_nonblock(sockets,  # type: Dict[Any, str]
+                     remain=None,  # type: Optional[Any]
+                     ):
+    # type: (...) -> Tuple[Dict[Any, str], Callable]
     """This function is called during sendrecv() routine to select
     the available sockets.
     """
     # pcap sockets aren't selectable, so we return all of them
     # and ask the selecting functions to use nonblock_recv instead of recv
     def _sleep_nonblock_recv(self):
+        # type: (Any) -> Any
         res = self.nonblock_recv()
         if res is None:
             time.sleep(conf.recv_poll_rate)