Fix AsyncSniffer not stopping promptly when timeout is set

Always add a control ObjectPipe so stop() can wake select() on
nonblocking backends (e.g. BPF on macOS). Previously the pipe was
only used for blocking sockets, leaving the sniff thread blocked for
the full timeout after stop(join=False).

Fixes #4890

AI-Assisted: yes [Cursor]
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: T3pp31

scapy/sendrecv.py

@@ -1286,32 +1286,22 @@ def _run(self,
         # Get select information from the sockets
         _main_socket = next(iter(sniff_sockets))
         select_func = _main_socket.select
-        nonblocking_socket = getattr(_main_socket, "nonblocking_socket", False)
         # We check that all sockets use the same select(), or raise a warning
         if not all(select_func == sock.select for sock in sniff_sockets):
             warning("Warning: inconsistent socket types ! "
                     "The used select function "
                     "will be the one of the first socket")
 
-        close_pipe = None  # type: Optional[ObjectPipe[None]]
-        if not nonblocking_socket:
-            # select is blocking: Add special control socket
-            from scapy.automaton import ObjectPipe
-            close_pipe = ObjectPipe[None]("control_socket")
-            sniff_sockets[close_pipe] = "control_socket"  # type: ignore
-
-            def stop_cb():
-                # type: () -> None
-                if self.running and close_pipe:
-                    close_pipe.send(None)
-                self.continue_sniff = False
-            self.stop_cb = stop_cb
-        else:
-            # select is non blocking
-            def stop_cb():
-                # type: () -> None
-                self.continue_sniff = False
-            self.stop_cb = stop_cb
+        from scapy.automaton import ObjectPipe
+        close_pipe = ObjectPipe[None]("control_socket")
+        sniff_sockets[close_pipe] = "control_socket"  # type: ignore
+
+        def stop_cb():
+            # type: () -> None
+            if self.running and close_pipe:
+                close_pipe.send(None)
+            self.continue_sniff = False
+        self.stop_cb = stop_cb
 
         try:
             self.continue_sniff = True

test/regression.uts

@@ -1782,6 +1782,43 @@ try:
 except ValueError:
     assert True
 
+= AsyncSniffer early stop with timeout on nonblocking socket (#4890)
+
+import select as select_mod
+import time
+from scapy.automaton import ObjectPipe
+from scapy.supersocket import SuperSocket
+
+class NonblockingSelectSocket(SuperSocket):
+    nonblocking_socket = True
+    def __init__(self):
+        self.ins = ObjectPipe(name="dummy_sniff_socket")
+        self.outs = None
+    def recv(self, x=65535, **kwargs):
+        self.ins.recv()
+        return None
+    @staticmethod
+    def select(sockets, remain=None):
+        if remain is None:
+            remain = 0.05
+        ready, unused_wr, unused_ex = select_mod.select(sockets, [], [], remain)
+        return ready
+    def close(self):
+        if self.closed:
+            return
+        self.closed = True
+        self.ins.close()
+
+sock = NonblockingSelectSocket()
+sniffer = AsyncSniffer(opened_socket=sock, timeout=10, count=0)
+sniffer.start()
+time.sleep(0.2)
+sniffer.stop(join=False)
+sniffer.thread.join(timeout=1)
+assert not sniffer.running
+assert not sniffer.thread.is_alive()
+sock.close()
+
 = Sending a TCP syn 'forever' at layer 2 and layer 3
 ~ netaccess needs_root IP
 def _test():