From 0731975a03631e4355d42e1714438b3fab7f8e5c Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Fri, 9 Jan 2026 21:20:01 +0100 Subject: [PATCH 1/2] Enhance SuperSocket for AnsweringMachine typing and compatibility improvements --- scapy/supersocket.py | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/scapy/supersocket.py b/scapy/supersocket.py index 5ffd7a30f63..0fdbdd6c21c 100644 --- a/scapy/supersocket.py +++ b/scapy/supersocket.py @@ -14,6 +14,7 @@ import struct import time +from scapy.ansmachine import AnsweringMachine from scapy.config import conf from scapy.consts import DARWIN, WINDOWS from scapy.data import ( @@ -44,14 +45,16 @@ Optional, Tuple, Type, + TypeVar, cast, ) +from scapy.compat import Self # Utils class _SuperSocket_metaclass(type): - desc = None # type: Optional[str] + desc = None # type: Optional[str] def __repr__(self): # type: () -> str @@ -82,10 +85,13 @@ class tpacket_auxdata(ctypes.Structure): # SuperSocket +_T = TypeVar("_T", Packet, PacketList) + + class SuperSocket(metaclass=_SuperSocket_metaclass): closed = False # type: bool nonblocking_socket = False # type: bool - auxdata_available = False # type: bool + auxdata_available = False # type: bool def __init__(self, family=socket.AF_INET, # type: int @@ -271,19 +277,17 @@ def tshark(self, *args, **kargs): from scapy import sendrecv sendrecv.tshark(opened_socket=self, *args, **kargs) - # TODO: use 'scapy.ansmachine.AnsweringMachine' when typed def am(self, - cls, # type: Type[Any] - *args, # type: Any + cls, # type: Type[AnsweringMachine[_T]] **kwargs # type: Any ): - # type: (...) -> Any + # type: (...) -> AnsweringMachine[_T] """ Creates an AnsweringMachine associated with this socket. :param cls: A subclass of AnsweringMachine to instantiate """ - return cls(*args, opened_socket=self, socket=self, **kwargs) + return cls(opened_socket=self, socket=self, **kwargs) @staticmethod def select(sockets, remain=conf.recv_poll_rate): @@ -295,6 +299,7 @@ def select(sockets, remain=conf.recv_poll_rate): :returns: an array of sockets that were selected and the function to be called next to get the packets (i.g. recv) """ + inp = [] # type: List[SuperSocket] try: inp, _, _ = select(sockets, [], [], remain) except (IOError, select_error) as exc: @@ -309,7 +314,7 @@ def __del__(self): self.close() def __enter__(self): - # type: () -> SuperSocket + # type: () -> Self return self def __exit__(self, exc_type, exc_value, traceback): @@ -627,6 +632,7 @@ def _iter(obj=cast(SndRcvList, obj)): s.time = s.sent_time yield s yield r + self.iter = _iter() elif isinstance(obj, (list, PacketList)): if isinstance(obj[0], bytes): From 08b6fa472164652bfb4669b9dfae9752c9985e57 Mon Sep 17 00:00:00 2001 From: Nils Weiss Date: Sat, 10 Jan 2026 19:57:58 +0100 Subject: [PATCH 2/2] Refactor imports in supersocket.py to use TYPE_CHECKING for AnsweringMachine. --- scapy/supersocket.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/scapy/supersocket.py b/scapy/supersocket.py index 0fdbdd6c21c..9c5b74a3b52 100644 --- a/scapy/supersocket.py +++ b/scapy/supersocket.py @@ -14,7 +14,6 @@ import struct import time -from scapy.ansmachine import AnsweringMachine from scapy.config import conf from scapy.consts import DARWIN, WINDOWS from scapy.data import ( @@ -47,9 +46,14 @@ Type, TypeVar, cast, + TYPE_CHECKING, ) from scapy.compat import Self +if TYPE_CHECKING: + from scapy.ansmachine import AnsweringMachine + + # Utils