diff --git a/scapy/arch/common.py b/scapy/arch/common.py index 2935e4280ba..09739be2ac7 100644 --- a/scapy/arch/common.py +++ b/scapy/arch/common.py @@ -7,7 +7,6 @@ Functions common to different architectures """ -import ctypes import re import socket @@ -15,7 +14,6 @@ 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)) diff --git a/scapy/arch/linux/__init__.py b/scapy/arch/linux/__init__.py index 29ad12cddc9..aa8cf05ff1a 100644 --- a/scapy/arch/linux/__init__.py +++ b/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 diff --git a/scapy/automaton.py b/scapy/automaton.py index 9c4ac93d35e..aaa67623266 100644 --- a/scapy/automaton.py +++ b/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 diff --git a/scapy/layers/bluetooth.py b/scapy/layers/bluetooth.py index 938e82e693f..55f580df5d4 100644 --- a/scapy/layers/bluetooth.py +++ b/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")) diff --git a/scapy/supersocket.py b/scapy/supersocket.py index 7b61ae3fbf2..16967e6fa37 100644 --- a/scapy/supersocket.py +++ b/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 \ diff --git a/test/regression.uts b/test/regression.uts index 1807df69758..5c38205744f 100644 --- a/test/regression.uts +++ b/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