def get_dev_bpf():
    """Returns an opened BPF file object"""

    # Get the first available BPF handle
    for bpf in range(256):
        try:
            fd = os.open("/dev/bpf%i" % bpf, os.O_RDWR)
            return (fd, bpf)
        except OSError as ex:
            if ex.errno == 13:  # Permission denied
                raise Scapy_Exception((
                    "Permission denied: could not open /dev/bpf%i. "
                    "Make sure to be running Scapy as root ! (sudo)"
                ) % bpf)
            continue

    raise Scapy_Exception("No /dev/bpf handle is available !") def attach_filter(fd, bpf_filter, iface):
    """Attach a BPF filter to the BPF file descriptor"""
    bp = compile_filter(bpf_filter, iface)
    # Assign the BPF program to the interface
    ret = LIBC.ioctl(c_int(fd), BIOCSETF, cast(pointer(bp), c_char_p))
    if ret < 0:
        raise Scapy_Exception("Can't attach the BPF filter !") BIOCGBLEN = _IOR('B', 102, ctypes.c_uint) BIOCGDLT = _IOR('B', 106, ctypes.c_uint) BIOCGSTATS = _IOR('B', 111, _bpf_stat) BIOCIMMEDIATE = _IOW('B', 112, ctypes.c_uint) BIOCPROMISC = _IO('B', 105) BIOCSBLEN = _IOWR('B', 102, ctypes.c_uint) BIOCSDLT = _IOW('B', 120, ctypes.c_uint) BIOCSETIF = _IOW('B', 108, 32) BIOCSHDRCMPLT = _IOW('B', 117, ctypes.c_uint) BIOCSTSTAMP = _IOW('B', 132, ctypes.c_uint) BPF_BUFFER_LENGTH = MTU BPF_T_NANOTIME = 0x0001 class ConfClass(object):
class Interceptor(object):
class ProgPath(ConfClass):
class ConfigFieldList:
class Emphasize(ConfigFieldList):
class Resolve(ConfigFieldList):
class Num2Layer:
class LayersList(List[Type['scapy.packet.Packet']]):
class CommandsList(List[Callable[..., Any]]):
class CacheInstance(Dict[str, Any], object):
class NetCache:
class Conf(ConfClass):
    def configure(self, cnf):
    def __repr__(self):
    def __str__(self):
    def __init__(self,
                 name,  # type: str
                 default,  # type: Any
                 hook,  # type: Callable[..., Any]
                 args=None,  # type: Optional[List[Any]]
                 kargs=None  # type: Optional[Dict[str, Any]]
                 ):
    def __get__(self, obj, typ=None):
    def set_from_hook(obj, name, val):
    def __set__(self, obj, val):
def _readonly(name):
    def __init__(self):
    def _is_field(f):
    def _recalc_layer_list(self):
    def add(self, *flds):
    def remove(self, *flds):
    def __contains__(self, elt):
    def __repr__(self):
    def __init__(self):
    def register(self, num, layer):
    def register_num2layer(self, num, layer):
    def register_layer2num(self, num, layer):
    def __getitem__(self, item):
    def __contains__(self, item):
    def get(self,
            item,  # type: Union[int, Type[Packet]]
            default=None,  # type: Optional[Type[Packet]]
            ):
    def __repr__(self):
    def __init__(self):
    def __repr__(self):
    def register(self, layer):
    def layers(self):
    def filter(self, items):
    def unfilter(self):
    def __repr__(self):
    def register(self, cmd):
def lsc():
    def __init__(self, name="noname", timeout=None):
    def flush(self):
    def __getitem__(self, item):
    def get(self, item, default=None):
    def __setitem__(self, item, v):
    def update(self,  # type: ignore
               other,  # type: Any
               **kwargs  # type: Any
               ):
    def iteritems(self):
    def iterkeys(self):
    def __iter__(self):
    def itervalues(self):
    def items(self):
    def keys(self):
    def values(self):
    def __len__(self):
    def summary(self):
    def __repr__(self):
    def copy(self):
    def __init__(self):
    def add_cache(self, cache):
    def new_cache(self, name, timeout=None):
    def __delattr__(self, attr):
    def update(self, other):
    def flush(self):
    def __repr__(self):
def _version_checker(module, minver):
def isCryptographyValid():
def isCryptographyAdvanced():
def isPyPy():
def _prompt_changer(attr, val, old):
def _set_conf_sockets():
def _socket_changer(attr, val, old):
def _loglevel_changer(attr, val, old):
def _iface_changer(attr, val, old):
    def __getattribute__(self, attr):
def crypto_validator(func):
    def func_in(*args, **kwargs):
def scapy_delete_temp_files(): class SuperSocket:
    closed = 0    # type: int
    nonblocking_socket = False  # type: bool
    auxdata_available = False   # type: bool

    def __init__(self,
                 family=socket.AF_INET,  # type: int
                 type=socket.SOCK_STREAM,  # type: int
                 proto=0,  # type: int
                 iface=None,  # type: Optional[_GlobInterfaceType]
                 **kwargs  # type: Any
                 ):
        # type: (...) -> None
        self.ins = socket.socket(family, type, proto)  # type: socket.socket
        self.outs = self.ins  # type: Optional[socket.socket]
        self.promisc = None
        self.iface = iface

    def send(self, x):
        # type: (Packet) -> int
        sx = raw(x)
        try:
            x.sent_time = time.time()
        except AttributeError:
            pass

        if self.outs:
            return self.outs.send(sx)
        else:
            return 0

    if six.PY2:
        def _recv_raw(self, sock, x):
            # type: (socket.socket, int) -> Tuple[bytes, Any, Optional[float]]
            """Internal function to receive a Packet"""
            pkt, sa_ll = sock.recvfrom(x)
            return pkt, sa_ll, None
    else:
        def _recv_raw(self, sock, x):
            # type: (socket.socket, int) -> Tuple[bytes, Any, Optional[float]]
            """Internal function to receive a Packet,
            and process ancillary data.
            """
            timestamp = None
            if not self.auxdata_available:
                pkt, _, _, sa_ll = sock.recvmsg(x)
                return pkt, sa_ll, timestamp
            flags_len = socket.CMSG_LEN(4096)
            pkt, ancdata, flags, sa_ll = sock.recvmsg(x, flags_len)
            if not pkt:
                return pkt, sa_ll, timestamp
            for cmsg_lvl, cmsg_type, cmsg_data in ancdata:
                # Check available ancillary data
                if (cmsg_lvl == SOL_PACKET and cmsg_type == PACKET_AUXDATA):
                    # Parse AUXDATA
                    try:
                        auxdata = tpacket_auxdata.from_buffer_copy(cmsg_data)
                    except ValueError:
                        # Note: according to Python documentation, recvmsg()
                        #       can return a truncated message. A ValueError
                        #       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:
                        # Insert VLAN tag
                        tpid = ETH_P_8021Q
                        if auxdata.tp_status & TP_STATUS_VLAN_TPID_VALID:
                            tpid = auxdata.tp_vlan_tpid
                        tag = struct.pack(
                            "!HH",
                            tpid,
                            auxdata.tp_vlan_tci
                        )
                        pkt = pkt[:12] + tag + pkt[12:]
                elif cmsg_lvl == socket.SOL_SOCKET and \
                        cmsg_type == SO_TIMESTAMPNS:
                    length = len(cmsg_data)
                    if length == 16:  # __kernel_timespec
                        tmp = struct.unpack("ll", cmsg_data)
                    elif length == 8:  # timespec
                        tmp = struct.unpack("ii", cmsg_data)
                    else:
                        log_runtime.warning("Unknown timespec format.. ?!")
                        continue
                    timestamp = tmp[0] + tmp[1] * 1e-9
            return pkt, sa_ll, timestamp

    def recv_raw(self, x=MTU):
        # type: (int) -> Tuple[Optional[Type[Packet]], Optional[bytes], Optional[float]]  # noqa: E501
        """Returns a tuple containing (cls, pkt_data, time)"""
        return conf.raw_layer, self.ins.recv(x), None

    def recv(self, x=MTU):
        # type: (int) -> Optional[Packet]
        cls, val, ts = self.recv_raw(x)
        if not val or not cls:
            return None
        try:
            pkt = cls(val)  # type: Packet
        except KeyboardInterrupt:
            raise
        except Exception:
            if conf.debug_dissector:
                from scapy.sendrecv import debug
                debug.crashed_on = (cls, val)
                raise
            pkt = conf.raw_layer(val)
        if ts:
            pkt.time = ts
        return pkt

    def fileno(self):
        # type: () -> int
        return self.ins.fileno()

    def close(self):
        # type: () -> None
        if self.closed:
            return
        self.closed = True
        if getattr(self, "outs", None):
            if getattr(self, "ins", None) != self.outs:
                if self.outs and (WINDOWS or self.outs.fileno() != -1):
                    self.outs.close()
        if getattr(self, "ins", None):
            if WINDOWS or self.ins.fileno() != -1:
                self.ins.close()

    def sr(self, *args, **kargs):
        # type: (Any, Any) -> Tuple[SndRcvList, PacketList]
        from scapy import sendrecv
        ans, unans = sendrecv.sndrcv(self, *args, **kargs)  # type: SndRcvList, PacketList  # noqa: E501
        return ans, unans

    def sr1(self, *args, **kargs):
        # type: (Any, Any) -> Optional[Packet]
        from scapy import sendrecv
        ans = sendrecv.sndrcv(self, *args, **kargs)[0]  # type: SndRcvList
        if len(ans) > 0:
            pkt = ans[0][1]  # type: Packet
            return pkt
        else:
            return None

    def sniff(self, *args, **kargs):
        # type: (Any, Any) -> PacketList
        from scapy import sendrecv
        pkts = sendrecv.sniff(opened_socket=self, *args, **kargs)  # type: PacketList  # noqa: E501
        return pkts

    def tshark(self, *args, **kargs):
        # type: (Any, Any) -> None
        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
           **kwargs  # type: Any
           ):
        # type: (...) -> Any
        """
        Creates an AnsweringMachine associated with this socket.

        :param cls: A subclass of AnsweringMachine to instantiate
        """
        return cls(*args, opened_socket=self, socket=self, **kwargs)

    @staticmethod
    def select(sockets, remain=conf.recv_poll_rate):
        # type: (List[SuperSocket], Optional[float]) -> List[SuperSocket]
        """This function is called during sendrecv() routine to select
        the available sockets.

        :param sockets: an array of sockets that need to be selected
        :returns: an array of sockets that were selected and
            the function to be called next to get the packets (i.g. recv)
        """
        try:
            inp, _, _ = select(sockets, [], [], remain)
        except (IOError, select_error) as exc:
            # select.error has no .errno attribute
            if not exc.args or exc.args[0] != errno.EINTR:
                raise
        return inp

    def __del__(self):
        # type: () -> None
        """Close the socket"""
        self.close()

    def __enter__(self):
        # type: () -> SuperSocket
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[Any]) -> None  # noqa: E501
        """Close the socket"""
        self.close() def raw(x):
    # type: (Union[Packet]) -> bytes
    """
    Builds a packet and returns its bytes representation.
    This function is and will always be cross-version compatible
    """
    if hasattr(x, "__bytes__"):
        return x.__bytes__()
    return bytes(x)
from select import select
from scapy.arch.bpf.core import get_dev_bpf, attach_filter
from scapy.arch.bpf.consts import (
    BIOCGBLEN,
    BIOCGDLT,
    BIOCGSTATS,
    BIOCIMMEDIATE,
    BIOCPROMISC,
    BIOCSBLEN,
    BIOCSDLT,
    BIOCSETIF,
    BIOCSHDRCMPLT,
    BIOCSTSTAMP,
    BPF_BUFFER_LENGTH,
    BPF_T_NANOTIME,
)
from scapy.config import conf
from scapy.consts import DARWIN, FREEBSD, NETBSD
from scapy.data import ETH_P_ALL, DLT_IEEE802_11_RADIO
from scapy.error import Scapy_Exception, warning
from scapy.interfaces import network_name
from scapy.supersocket import SuperSocket
from scapy.compat import raw
        from scapy.layers.l2 import Loopback
import ctypes
import errno
import fcntl
import os
import platform
import struct
import sys
import time
            "Can't send anything with %s" % self.__class__.__name__
        )

    def recv_raw(self, x=BPF_BUFFER_LENGTH):
        """Dummy recv method"""
        raise Exception(
            "Can't recv anything with %s" % self.__class__.__name__
        )

    @staticmethod
    def select(sockets, remain=None):
        """This function is called during sendrecv() routine to select
        the available sockets.
        """
        # sockets, None (means use the socket's recv() )
        return bpf_select(sockets, remain)


class L2bpfListenSocket(_L2bpfSocket):
    """"Scapy L2 BPF Listen Super Socket"""

    def __init__(self, *args, **kwargs):
        self.received_frames = []
        super(L2bpfListenSocket, self).__init__(*args, **kwargs)

    def buffered_frames(self):
        """Return the number of frames in the buffer"""
        return len(self.received_frames)

    def get_frame(self):
        """Get a frame or packet from the received list"""
        if self.received_frames:
            return self.received_frames.pop(0)
        else:
            return None, None, None

    @staticmethod
    def bpf_align(bh_h, bh_c):
        """Return the index to the end of the current packet"""

        # from <net/bpf.h>
        return ((bh_h + bh_c) + (BPF_ALIGNMENT - 1)) & ~(BPF_ALIGNMENT - 1)

    def extract_frames(self, bpf_buffer):
        """
        Extract all frames from the buffer and stored them in the received list
        """

        # Ensure that the BPF buffer contains at least the header
        len_bb = len(bpf_buffer)
        if len_bb < _bpf_hdr_len:
            return

        # Extract useful information from the BPF header
        bh_hdr = bpf_hdr.from_buffer_copy(bpf_buffer)
        if bh_hdr.bh_datalen == 0:
            return

        # Get and store the Scapy object
        frame_str = bpf_buffer[
            bh_hdr.bh_hdrlen:bh_hdr.bh_hdrlen + bh_hdr.bh_caplen
        ]
        if _NANOTIME:
            ts = bh_hdr.bh_tstamp.tv_sec + 1e-9 * bh_hdr.bh_tstamp.tv_nsec
        else:
            ts = bh_hdr.bh_tstamp.tv_sec + 1e-6 * bh_hdr.bh_tstamp.tv_usec
        self.received_frames.append(
            (self.guessed_cls, frame_str, ts)
        )

        # Extract the next frame
        end = self.bpf_align(bh_hdr.bh_hdrlen, bh_hdr.bh_caplen)
        if (len_bb - end) >= 20:
            self.extract_frames(bpf_buffer[end:])

    def recv_raw(self, x=BPF_BUFFER_LENGTH):
        """Receive a frame from the network"""

        x = min(x, BPF_BUFFER_LENGTH)

        if self.buffered_frames():
            # Get a frame from the buffer
            return self.get_frame()

        # Get data from BPF
        try:
            bpf_buffer = os.read(self.ins, x)
        except EnvironmentError as exc:
            if exc.errno != errno.EAGAIN:
                warning("BPF recv_raw()", exc_info=True)
            return None, None, None

        # Extract all frames from the BPF buffer
        self.extract_frames(bpf_buffer)
        return self.get_frame()


class L2bpfSocket(L2bpfListenSocket):
    """"Scapy L2 BPF Super Socket"""

    def send(self, x):
        """Send a frame"""
        return os.write(self.outs, raw(x))

    def nonblock_recv(self):
        """Non blocking receive"""

        if self.buffered_frames():
            # Get a frame from the buffer
            return L2bpfListenSocket.recv(self)

        # Set the non blocking flag, read from the socket, and unset the flag
        self.set_nonblock(True)
        pkt = L2bpfListenSocket.recv(self)
        self.set_nonblock(False)
        return pkt


class L3bpfSocket(L2bpfSocket):


    def recv(self, x=BPF_BUFFER_LENGTH):