Defer ctypes-dependent startup imports

scapy/arch/common.py

@@ -7,15 +7,13 @@
 Functions common to different architectures
 """
 
-import ctypes
 import re
 import socket
 
 from scapy.config import conf
 from scapy.data import MTU, ARPHRD_TO_DLT, DLT_RAW_ALT, DLT_RAW
 from scapy.error import Scapy_Exception, warning
 from scapy.interfaces import network_name, resolve_iface, NetworkInterface
-from scapy.libs.structures import bpf_program
 from scapy.pton_ntop import inet_pton
 from scapy.utils import decode_locale_str
 
@@ -24,9 +22,13 @@
 from typing import (
     List,
     Optional,
+    TYPE_CHECKING,
     Union,
 )
 
+if TYPE_CHECKING:
+    from scapy.libs.structures import bpf_program
+
 # From if.h
 _iff_flags = [
     "UP",
@@ -68,13 +70,20 @@ def compile_filter(filter_exp,  # type: str
                    linktype=None,  # type: Optional[int]
                    promisc=False  # type: bool
                    ):
-    # type: (...) -> bpf_program
+    # type: (...) -> 'bpf_program'
     """Asks libpcap to parse the filter, then build the matching
     BPF bytecode.
 
     :param iface: if provided, use the interface to compile
     :param linktype: if provided, use the linktype to compile
     """
+    try:
+        import ctypes
+    except OSError:
+        raise ImportError(
+            "ctypes is not available. Cannot compile filter !"
+        )
+    from scapy.libs.structures import bpf_program
     try:
         from scapy.libs.winpcapy import (
             PCAP_ERRBUF_SIZE,
@@ -141,11 +150,22 @@ def compile_filter(filter_exp,  # type: str
     return bpf
 
 
-def free_filter(bp: bpf_program) -> None:
+def free_filter(bp: 'bpf_program') -> None:
     """
     Free a bpf_program created with compile_filter
     """
-    from scapy.libs.winpcapy import pcap_freecode
+    try:
+        import ctypes
+    except OSError:
+        raise ImportError(
+            "ctypes is not available. Cannot free filter !"
+        )
+    try:
+        from scapy.libs.winpcapy import pcap_freecode
+    except OSError:
+        raise ImportError(
+            "libpcap is not available. Cannot free filter !"
+        )
     pcap_freecode(ctypes.byref(bp))
 
 

scapy/arch/linux/__init__.py

@@ -11,7 +11,6 @@
 from fcntl import ioctl
 from select import select
 
-import ctypes
 import os
 import socket
 import struct
@@ -38,7 +37,6 @@
     network_name,
     resolve_iface,
 )
-from scapy.libs.structures import sock_fprog
 from scapy.packet import Packet, Padding
 from scapy.supersocket import SuperSocket
 
@@ -126,6 +124,12 @@ def attach_filter(sock, bpf_filter, iface):
     :param bpf_filter: the bpf string filter to compile
     :param iface: the interface used to compile
     """
+    try:
+        import ctypes
+        from scapy.libs.structures import sock_fprog
+    except OSError:
+        raise Scapy_Exception("ctypes is not available")
+
     bp = compile_filter(bpf_filter, iface)
     if conf.use_pypy and sys.pypy_version_info <= (7, 3, 2):  # type: ignore
         # PyPy < 7.3.2 has a broken behavior

scapy/automaton.py

@@ -11,7 +11,6 @@
     - add documentation for ioevent, as_supersocket...
 """
 
-import ctypes
 import itertools
 import logging
 import os
@@ -56,6 +55,11 @@
 )
 from scapy.compat import DecoratorCallable
 
+ctypes = cast(Any, None)
+if WINDOWS:
+    import ctypes as _ctypes
+    ctypes = _ctypes
+
 
 # winsock.h
 FD_READ = 0x00000001

scapy/layers/bluetooth.py

@@ -10,11 +10,16 @@
 Bluetooth layers, sockets and send/receive functions.
 """
 
-import ctypes
 import socket
 import struct
 import select
-from ctypes import sizeof
+
+try:
+    import ctypes
+    from ctypes import sizeof
+except OSError:
+    ctypes = None
+    sizeof = None
 
 from scapy.config import conf
 from scapy.data import (
@@ -64,7 +69,7 @@
 from scapy.sendrecv import sndrcv
 from scapy.data import MTU
 from scapy.consts import WINDOWS
-from scapy.error import warning
+from scapy.error import Scapy_Exception, warning
 
 
 ############
@@ -3576,12 +3581,16 @@ def recv(self, x=MTU):
         return HCI_Hdr(self.ins.recv(x))
 
 
-class sockaddr_hci(ctypes.Structure):
-    _fields_ = [
-        ("sin_family", ctypes.c_ushort),
-        ("hci_dev", ctypes.c_ushort),
-        ("hci_channel", ctypes.c_ushort),
-    ]
+if ctypes is not None:
+    class sockaddr_hci(ctypes.Structure):
+        _fields_ = [
+            ("sin_family", ctypes.c_ushort),
+            ("hci_dev", ctypes.c_ushort),
+            ("hci_channel", ctypes.c_ushort),
+        ]
+else:
+    class sockaddr_hci(object):
+        pass
 
 
 class _BluetoothLibcSocket(SuperSocket):
@@ -3593,6 +3602,8 @@ def __init__(self, socket_domain, socket_type, socket_protocol, sock_address):
         # Python socket and bind implementations do not allow us to pass down
         # the correct parameters. We must call libc functions directly via
         # ctypes.
+        if ctypes is None:
+            raise Scapy_Exception("ctypes is not available")
         sockaddr_hcip = ctypes.POINTER(sockaddr_hci)
         from ctypes.util import find_library
         libc = ctypes.cdll.LoadLibrary(find_library("c"))

scapy/supersocket.py

@@ -8,7 +8,6 @@
 """
 
 from select import select, error as select_error
-import ctypes
 import errno
 import socket
 import struct
@@ -75,18 +74,6 @@ def __repr__(self):
 TP_STATUS_VLAN_TPID_VALID = 1 << 6
 
 
-class tpacket_auxdata(ctypes.Structure):
-    _fields_ = [
-        ("tp_status", ctypes.c_uint),
-        ("tp_len", ctypes.c_uint),
-        ("tp_snaplen", ctypes.c_uint),
-        ("tp_mac", ctypes.c_ushort),
-        ("tp_net", ctypes.c_ushort),
-        ("tp_vlan_tci", ctypes.c_ushort),
-        ("tp_vlan_tpid", ctypes.c_ushort),
-    ]  # type: List[Tuple[str, Any]]
-
-
 # SuperSocket
 
 _T = TypeVar("_T", Packet, PacketList)
@@ -162,23 +149,26 @@ def _recv_raw(self, sock, x):
                 if (cmsg_lvl == SOL_PACKET and cmsg_type == PACKET_AUXDATA):
                     # Parse AUXDATA
                     try:
-                        auxdata = tpacket_auxdata.from_buffer_copy(cmsg_data)
-                    except ValueError:
+                        # tpacket_auxdata fields, in order:
+                        # tp_status, tp_len, tp_snaplen, tp_mac, tp_net,
+                        # tp_vlan_tci, tp_vlan_tpid
+                        tp_status, _, _, _, _, tp_vlan_tci, tp_vlan_tpid = \
+                            struct.unpack("=IIIHHHH", cmsg_data[:20])
+                    except struct.error:
                         # Note: according to Python documentation, recvmsg()
-                        #       can return a truncated message. A ValueError
+                        #       can return a truncated message. A struct.error
                         #       exception likely indicates that Auxiliary
                         #       Data is not supported by the Linux kernel.
                         return pkt, sa_ll, timestamp
-                    if auxdata.tp_vlan_tci != 0 or \
-                            auxdata.tp_status & TP_STATUS_VLAN_VALID:
+                    if tp_vlan_tci != 0 or tp_status & TP_STATUS_VLAN_VALID:
                         # Insert VLAN tag
                         tpid = ETH_P_8021Q
-                        if auxdata.tp_status & TP_STATUS_VLAN_TPID_VALID:
-                            tpid = auxdata.tp_vlan_tpid
+                        if tp_status & TP_STATUS_VLAN_TPID_VALID:
+                            tpid = tp_vlan_tpid
                         tag = struct.pack(
                             "!HH",
                             tpid,
-                            auxdata.tp_vlan_tci
+                            tp_vlan_tci
                         )
                         pkt = pkt[:12] + tag + pkt[12:]
                 elif cmsg_lvl == socket.SOL_SOCKET and \

test/regression.uts

@@ -759,6 +759,41 @@ except:
 
 interact_emulator(extra_args=["-d"])  # Extended
 
+= init_session() without ctypes
+~ interact
+
+import subprocess
+import sys
+import textwrap
+
+code = textwrap.dedent("""
+    import builtins
+
+    from scapy.main import init_session
+
+    real_import = builtins.__import__
+
+    def fake_import(name, globals=None, locals=None, fromlist=(), level=0):
+        if name == "ctypes" and globals and globals.get("__name__") == "scapy.supersocket":
+            raise OSError("Dynamic loading not supported")
+        return real_import(name, globals, locals, fromlist, level)
+
+    builtins.__import__ = fake_import
+    try:
+        ns = init_session(ret=True)
+    finally:
+        builtins.__import__ = real_import
+
+    assert "Ether" in ns
+""")
+
+result = subprocess.run(
+    [sys.executable, "-c", code],
+    capture_output=True,
+    text=True,
+)
+assert result.returncode == 0, result.stderr
+
 = Emulate interact() and test startup.py with ptpython
 ~ interact