Sendrecv overhaul: async sniffing & major cleanup

.travis.yml

@@ -134,6 +134,8 @@ matrix:
           env:
             - TOXENV=linux_warnings
 
-install: bash .travis/install.sh
+install:
+  - bash .travis/install.sh
+  - python -c "from scapy.all import conf; print(repr(conf))"
 
 script: bash .travis/test.sh

doc/scapy/usage.rst

@@ -698,6 +698,52 @@ We can sniff and do passive OS fingerprinting::
 
 The number before the OS guess is the accuracy of the guess.
 
+Asynchronous Sniffing
+---------------------
+
+.. index::
+   single: AsyncSniffer()
+
+.. note::
+   Asynchronous sniffing is only available since **Scapy 2.4.3**
+
+It is possible to sniff asynchronously. This allows to stop the sniffer programmatically, rather than with ctrl^C.
+It provides ``start()``, ``stop()`` and ``join()`` utils.
+
+The basic usage would be:
+
+.. code-block:: python
+
+    >>> t = AsyncSniffer()
+    >>> t.start()
+    >>> print("hey")
+    hey
+    [...]
+    >>> results = t.stop()
+
+.. image:: graphics/animations/animation-scapy-asyncsniffer.svg
+
+The ``AsyncSniffer`` class has a few useful keys, such as ``results`` (the packets collected) or ``running``, that can be used.
+It accepts the same arguments than ``sniff()`` (in fact, their implementations are merged). For instance:
+
+.. code-block:: python
+
+    >>> t = AsyncSniffer(iface="enp0s3", count=200)
+    >>> t.start()
+    >>> t.join()  # this will hold until 200 packets are collected
+    >>> results = t.results
+    >>> print(len(results))
+    200
+
+Another example: using ``prn`` and ``store=False``
+
+.. code-block:: python
+
+    >>> t = AsyncSniffer(prn=lambda x: x.summary(), store=False, filter="tcp")
+    >>> t.start()
+    >>> time.sleep(20)
+    >>> t.stop()
+
 Advanced Sniffing - Sessions
 ----------------------------
 

scapy/arch/__init__.py

@@ -56,9 +56,9 @@ def get_if_hwaddr(iff):
     from scapy.arch.linux import *  # noqa F403
 elif BSD:
     from scapy.arch.unix import read_routes, read_routes6, in6_getifaddr  # noqa: F401, E501
-
-    if not conf.use_pcap or conf.use_dnet:
-        from scapy.arch.bpf.core import *  # noqa F403
+    from scapy.arch.bpf.core import *  # noqa F403
+    if not (conf.use_pcap or conf.use_dnet):
+        # Native
         from scapy.arch.bpf.supersocket import * # noqa F403
         conf.use_bpf = True
 elif SOLARIS:

scapy/arch/bpf/supersocket.py

@@ -35,13 +35,12 @@ class _L2bpfSocket(SuperSocket):
     """"Generic Scapy BPF Super Socket"""
 
     desc = "read/write packets using BPF"
-    assigned_interface = None
-    fd_flags = None
-    ins = None
-    closed = False
+    nonblocking_socket = True
 
     def __init__(self, iface=None, type=ETH_P_ALL, promisc=None, filter=None,
                  nofilter=0, monitor=False):
+        self.fd_flags = None
+        self.assigned_interface = None
 
         # SuperSocket mandatory variables
         if promisc is None:
@@ -208,11 +207,15 @@ def close(self):
 
     def send(self, x):
         """Dummy send method"""
-        raise Exception("Can't send anything with %s" % self.__name__)
+        raise Exception(
+            "Can't send anything with %s" % self.__class__.__name__
+        )
 
-    def recv(self, x=BPF_BUFFER_LENGTH):
+    def recv_raw(self, x=BPF_BUFFER_LENGTH):
         """Dummy recv method"""
-        raise Exception("Can't recv anything with %s" % self.__name__)
+        raise Exception(
+            "Can't recv anything with %s" % self.__class__.__name__
+        )
 
     @staticmethod
     def select(sockets, remain=None):
@@ -273,22 +276,20 @@ def extract_frames(self, bpf_buffer):
 
         # Get and store the Scapy object
         frame_str = bpf_buffer[bh_hdrlen:bh_hdrlen + bh_caplen]
-        try:
-            pkt = self.guessed_cls(frame_str)
-        except Exception:
-            if conf.debug_dissector:
-                raise
-            pkt = conf.raw_layer(frame_str)
-        self.received_frames.append(pkt)
+        self.received_frames.append(
+            (self.guessed_cls, frame_str, None)
+        )
 
         # Extract the next frame
         end = self.bpf_align(bh_hdrlen, bh_caplen)
         if (len_bb - end) >= 20:
             self.extract_frames(bpf_buffer[end:])
 
-    def recv(self, x=BPF_BUFFER_LENGTH):
+    def recv_raw(self, x=BPF_BUFFER_LENGTH):
         """Receive a frame from the network"""
 
+        x = min(x, BPF_BUFFER_LENGTH)
+
         if self.buffered_frames():
             # Get a frame from the buffer
             return self.get_frame()
@@ -299,7 +300,7 @@ def recv(self, x=BPF_BUFFER_LENGTH):
         except EnvironmentError as exc:
             if exc.errno != errno.EAGAIN:
                 warning("BPF recv()", exc_info=True)
-            return
+            return None, None, None
 
         # Extract all frames from the BPF buffer
         self.extract_frames(bpf_buffer)
@@ -318,7 +319,7 @@ def nonblock_recv(self):
 
         if self.buffered_frames():
             # Get a frame from the buffer
-            return self.get_frame()
+            return L2bpfListenSocket.recv(self)
 
         # Set the non blocking flag, read from the socket, and unset the flag
         self.set_nonblock(True)
@@ -329,11 +330,11 @@ def nonblock_recv(self):
 
 class L3bpfSocket(L2bpfSocket):
 
-    def get_frame(self):
-        """Get a frame or packet from the received list"""
-        pkt = super(L3bpfSocket, self).get_frame()
-        if pkt is not None:
-            return pkt.payload
+    def recv(self, x=BPF_BUFFER_LENGTH):
+        """Receive on layer 3"""
+        r = SuperSocket.recv(self, x)
+        if r:
+            return r.payload
 
     def send(self, pkt):
         """Send a packet"""
@@ -363,7 +364,10 @@ def send(self, pkt):
 
 def isBPFSocket(obj):
     """Return True is obj is a BPF Super Socket"""
-    return isinstance(obj, L2bpfListenSocket) or isinstance(obj, L2bpfListenSocket) or isinstance(obj, L3bpfSocket)  # noqa: E501
+    return isinstance(
+        obj,
+        (L2bpfListenSocket, L2bpfListenSocket, L3bpfSocket)
+    )
 
 
 def bpf_select(fds_list, timeout=None):

scapy/arch/common.py

@@ -63,24 +63,18 @@ def get_if(iff, cmd):
 
 # SOCKET UTILS
 
-class TimeoutElapsed(Scapy_Exception):
-    pass
-
-
 def _select_nonblock(sockets, remain=None):
     """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):
-        try:
-            res = self.nonblock_recv()
-            if res is None:
-                time.sleep(conf.recv_poll_rate)
-            return res
-        except TimeoutElapsed:
-            return None
+        res = self.nonblock_recv()
+        if res is None:
+            time.sleep(conf.recv_poll_rate)
+        return res
+    # we enforce remain=None: don't wait.
     return sockets, _sleep_nonblock_recv
 
 # BPF HANDLERS

scapy/arch/pcapdnet.py

@@ -14,7 +14,7 @@
 import time
 
 from scapy.automaton import SelectableObject
-from scapy.arch.common import _select_nonblock, TimeoutElapsed
+from scapy.arch.common import _select_nonblock
 from scapy.compat import raw, plain_str, chb
 from scapy.config import conf
 from scapy.consts import WINDOWS
@@ -38,7 +38,7 @@
 
 
 class _L2pcapdnetSocket(SuperSocket, SelectableObject):
-    read_allowed_exceptions = (TimeoutElapsed,)
+    nonblocking_socket = True
 
     def check_recv(self):
         return True
@@ -58,8 +58,6 @@ def recv_raw(self, x=MTU):
             pkt = self.ins.next()
             if pkt is not None:
                 ts, pkt = pkt
-            if pkt is None and scapy.consts.WINDOWS:
-                raise TimeoutElapsed  # To understand this behavior, have a look at L2pcapListenSocket's note  # noqa: E501
             if pkt is None:
                 return None, None, None
         return cls, pkt, ts
@@ -344,15 +342,11 @@ def send(self, x):
 
     class L3pcapSocket(L2pcapSocket):
         desc = "read/write packets at layer 3 using only libpcap"
-        # def __init__(self, iface = None, type = ETH_P_ALL, filter=None, nofilter=0):  # noqa: E501
-        #    L2pcapSocket.__init__(self, iface, type, filter, nofilter)
 
         def recv(self, x=MTU):
             r = L2pcapSocket.recv(self, x)
             if r:
                 return r.payload
-            else:
-                return
 
         def send(self, x):
             # Makes send detects when it should add Loopback(), Dot11... instead of Ether()  # noqa: E501

scapy/arch/windows/native.py

@@ -43,13 +43,14 @@
 the tests (windows.uts) for an example.
 """
 
+import io
 import os
 import socket
 import subprocess
 import time
 
 from scapy.automaton import SelectableObject
-from scapy.arch.common import _select_nonblock, TimeoutElapsed
+from scapy.arch.common import _select_nonblock
 from scapy.arch.windows.structures import GetIcmpStatistics
 from scapy.compat import raw
 from scapy.config import conf
@@ -62,7 +63,7 @@
 
 class L3WinSocket(SuperSocket, SelectableObject):
     desc = "a native Layer 3 (IPv4) raw socket under Windows"
-    read_allowed_exceptions = (TimeoutElapsed,)
+    nonblocking_socket = True
     __slots__ = ["promisc", "cls", "ipv6", "proto"]
 
     def __init__(self, iface=None, proto=socket.IPPROTO_IP,
@@ -114,7 +115,8 @@ def __init__(self, iface=None, proto=socket.IPPROTO_IP,
         self.ins.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
         self.outs.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
         # Bind on all ports
-        host = iface.ip if iface else socket.gethostname()
+        iface = iface or conf.iface
+        host = iface.ip if iface.ip else socket.gethostname()
         self.ins.bind((host, 0))
         self.ins.setblocking(False)
         # Get as much data as possible: reduce what is cropped
@@ -163,7 +165,7 @@ def nonblock_recv(self, x=MTU):
     def recv_raw(self, x=MTU):
         try:
             data, address = self.ins.recvfrom(x)
-        except IOError:  # BlockingIOError
+        except io.BlockingIOError:
             return None, None, None
         from scapy.layers.inet import IP
         from scapy.layers.inet6 import IPv6
@@ -190,7 +192,7 @@ def close(self):
 
     @staticmethod
     def select(sockets, remain=None):
-        return _select_nonblock(sockets, remain=None)
+        return _select_nonblock(sockets, remain=remain)
 
 
 class L3WinSocket6(L3WinSocket):

scapy/automaton.py

@@ -162,7 +162,7 @@ def process(self):
                     select_inputs.append(i)
                 elif not self.remain and i.check_recv():
                     self.results.append(i)
-                else:
+                elif self.remain:
                     i.wait_return(self._exit_door)
             if select_inputs:
                 # Use default select function
@@ -193,6 +193,8 @@ def select_objects(inputs, remain):
 
 
 class ObjectPipe(SelectableObject):
+    read_allowed_exceptions = ()
+
     def __init__(self):
         self.rd, self.wr = os.pipe()
         self.queue = deque()
@@ -218,6 +220,11 @@ def recv(self, n=0):
     def read(self, n=0):
         return self.recv(n)
 
+    def close(self):
+        os.close(self.rd)
+        os.close(self.wr)
+        self.queue.clear()
+
 
 class Message:
     def __init__(self, **args):

scapy/contrib/cansocket_native.py

@@ -28,6 +28,7 @@
 
 class CANSocket(SuperSocket):
     desc = "read/write packets at a given CAN interface using PF_CAN sockets"
+    nonblocking_socket = True
 
     def __init__(self, iface=None, receive_own_messages=False,
                  can_filters=None, remove_padding=True, basecls=CAN):

scapy/contrib/cansocket_python_can.py

@@ -29,6 +29,7 @@ class CANSocketTimeoutElapsed(Scapy_Exception):
 
 class CANSocket(SuperSocket):
     read_allowed_exceptions = (CANSocketTimeoutElapsed,)
+    nonblocking_socket = True
     desc = "read/write packets at a given CAN interface " \
            "using a python-can bus object"
 

scapy/contrib/isotp.py

@@ -533,6 +533,8 @@ class ISOTPSoftSocket(SuperSocket):
     * All background threads can be stopped by the garbage collector
     """
 
+    nonblocking_socket = True
+
     def __init__(self,
                  can_socket=None,
                  sid=0,

scapy/contrib/isotp.uts

@@ -23,6 +23,7 @@ iface1 = "vcan1"
 
 
 class MockCANSocket(SuperSocket):
+    nonblocking_socket = True
     def __init__(self, rcvd_queue=None):
         self.rcvd_queue = Queue()
         self.sent_queue = Queue()
@@ -737,9 +738,9 @@ with ISOTPSoftSocket(cans, sid=0x641, did=0x241) as s:
             raise ex
     thread = threading.Thread(target=sender, name="sender")
     thread.start()
-    ready.wait()
+    ready.wait(15)
     msg = s.recv()
-    thread.join()
+    thread.join(15)
 
 if exception is not None:
     raise exception
@@ -781,7 +782,7 @@ ready.wait()
 with ISOTPSoftSocket(cans, sid=0x641, did=0x241) as s:
     s.send(msg)
 
-thread.join()
+thread.join(15)
 succ.wait(2)
 assert(succ.is_set())