tests/sys/netpfil/common/sniffer.py

   1 # $FreeBSD$
   2
   3 import threading
   4 import scapy.all as sp
   5
   6 class Sniffer(threading.Thread):
   7         def __init__(self, args, check_function):
   8                 threading.Thread.__init__(self)
   9
  10                 self._args = args
  11                 self._recvif = args.recvif[0]
  12                 self._check_function = check_function
  13                 self.foundCorrectPacket = False
  14
  15                 self.start()
  16
  17         def _checkPacket(self, packet):
  18                 ret = self._check_function(self._args, packet)
  19                 if ret:
  20                         self.foundCorrectPacket = True
  21                 return ret
  22
  23         def run(self):
  24                 self.packets = sp.sniff(iface=self._recvif,
  25                                 stop_filter=self._checkPacket, timeout=3)