我发现了类似的问题:
(
Instance variables not being updated Python when using Multiprocessing),
但仍然不知道我的任务的解决方案.
任务是在testskript的完整性之后停止scapy sniff函数.单个测试脚本的运行持续时间可能会有很大差异(从几秒到几小时).我的嗅探功能在一个单独的威胁中运行. testscript在开始时调用init Funktion,从另一个模块调用sniff函数.
@classmethod
def SaveFullTrafficPcap(self, TestCase, Termination):
try:
Full_Traffic = []
PktList = []
FullPcapName = Settings['GeneralSettings']['ResultsPath']+TestCase.TestCaseName +"Full_Traffic_PCAP.pcap"
#while Term.Termination < 1:
Full_Traffic = sniff(lfilter=None, iface=str(Settings['GeneralSettings']['EthInterface']), store=True, prn = lambda x: Full_Traffic.append(x), count=0, timeout=Term.Termination)
print(Full_Traffic)
wrpcap(FullPcapName, Full_Traffic)
except(Exception):
SYS.ABS_print("No full traffic PCAP file wirtten!\n")
在testscript的末尾调用一个exit函数.在退出函数中,我将Term.Termination参数设置为1并等待5秒,但它不起作用. sniff函数由系统停止,我没有文件“FullPCAPName”
如果count或timeout获得一个值,代码可以正常工作,我得到我的FullPCAPName文件,并在我的界面上编译流量.
有没有人知道如何在完成测试后定期调整嗅探功能?
最佳答案 使用指定
here的stop_filter命令对我有用.为方便起见,我在下面重复了
HenningCash’s代码:
import time, threading
from scapy.all import sniff
e = threading.Event()
def _sniff(e):
a = sniff(filter="tcp port 80", stop_filter=lambda p: e.is_set())
print("Stopped after %i packets" % len(a))
print("Start capturing thread")
t = threading.Thread(target=_sniff, args=(e,))
t.start()
time.sleep(3)
print("Try to shutdown capturing...")
e.set()
# This will run until you send a HTTP request somewhere
# There is no way to exit clean if no package is received
while True:
t.join(2)
if t.is_alive():
print("Thread is still running...")
else:
break
print("Shutdown complete!")
但是,您仍然需要等待最终数据包被嗅探,这在您的方案中可能并不理想.