arch: minor cleanups to winpcapy and libpcap

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()

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
 
 

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:

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