import select
from scapy.all import ETH_P_IP, ETH_P_IPV6, ETH_P_ALL, ETH_P_ARP
import select
from scapy.all import ETH_P_IP, ETH_P_IPV6, ETH_P_ALL, ETH_P_ARP
from scapy.layers.inet6 import IPv6
__all__ = ['RxQueue', 'TxQueue', 'Interface', 'create_gratuitous_arp_request',
'auto_pad', 'checksum_equal']
from scapy.layers.inet6 import IPv6
__all__ = ['RxQueue', 'TxQueue', 'Interface', 'create_gratuitous_arp_request',
'auto_pad', 'checksum_equal']
:param queue: Queue in which this function will push incoming packets.
:type interface_name: str
:type queue: multiprocessing.Queue
:param queue: Queue in which this function will push incoming packets.
:type interface_name: str
:type queue: multiprocessing.Queue
"""
sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, ETH_P_ALL)
sock.bind((interface_name, ETH_P_ALL))
"""
sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, ETH_P_ALL)
sock.bind((interface_name, ETH_P_ALL))
:rtype: scapy.Ether
"""
(rlist, _, _) = select.select([self._sock], [], [], timeout)
:rtype: scapy.Ether
"""
(rlist, _, _) = select.select([self._sock], [], [], timeout)
self.sent_packets.append(pkt)
self.txq.send(pkt)
def recv_pkt(self, timeout=3):
self.sent_packets.append(pkt)
self.txq.send(pkt)
def recv_pkt(self, timeout=3):