使用Python捕获网络流量：解决嗅探与Signal消息发送并行问题

Scapy的sniff()是阻塞式调用，会占用主线程导致无法同时执行消息发送操作，核心解决思路是用多线程将嗅探任务放到后台线程，主线程负责发送Signal消息并控制嗅探启停，同时完成数据包捕获计时。

实现逻辑 #

• 用threading.Thread启动嗅探子线程，避免阻塞主线程
• 用线程安全的标志变量控制嗅探停止
• 在嗅探启动和结束时记录时间，计算总捕获时长
• 主线程完成Signal消息发送后，触发嗅探停止

完整代码示例 #

import threading
import time
from scapy.all import sniff, PacketList
import subprocess

# 全局控制变量
stop_sniffing = False
captured_packets = PacketList()
start_timestamp = 0.0
end_timestamp = 0.0

def sniff_traffic(interface="eth0"):
    global stop_sniffing, captured_packets, start_timestamp
    start_timestamp = time.time()
    
    # 自定义停止条件：每次捕获数据包时检查标志
    def stop_condition(pkt):
        return stop_sniffing
    
    # 启动嗅探，prn参数可选，用于实时打印数据包摘要
    captured_packets = sniff(iface=interface, stop_filter=stop_condition, prn=lambda pkt: print(pkt.summary()))

def send_signal_msg(target_num, content):
    # 调用signal-cli发送消息，需提前安装并配置好账号
    subprocess.run([
        "signal-cli", "-u", "你的Signal账号", "send", "-m", content, target_num
    ], check=True)

if __name__ == "__main__":
    # 1. 启动嗅探线程
    sniff_thread = threading.Thread(target=sniff_traffic, args=("eth0",))
    sniff_thread.start()
    
    # 短暂等待确保嗅探启动完成
    time.sleep(1)
    
    # 2. 发送Signal消息
    print("正在发送Signal消息...")
    send_signal_msg("目标手机号", "测试捕获触发消息")
    
    # 等待消息发送完成（可根据网络情况调整时长）
    time.sleep(2)
    
    # 3. 停止嗅探并等待线程结束
    stop_sniffing = True
    sniff_thread.join()
    
    # 4. 计算并输出计时结果
    end_timestamp = time.time()
    capture_duration = end_timestamp - start_timestamp
    print(f"捕获总时长: {capture_duration:.2f} 秒")
    print(f"捕获数据包总数: {len(captured_packets)} 个")
    
    # 可选：将捕获的数据包保存为pcap文件
    captured_packets.write("signal_traffic.pcap")

关键细节说明 #

• 停止机制：Scapy的sniff()支持stop_filter参数，每次捕获数据包都会执行该函数，返回True时立即停止嗅探，确保停止操作的及时性。
• Signal消息发送：示例使用signal-cli命令行工具，需提前完成安装和账号关联（绑定你的Signal设备）；也可替换为Python封装库（如pysignalclirestapi），但需部署对应服务。
• 接口选择：用scapy.all.get_if_list()可查看当前系统所有网络接口，将代码中的eth0替换为实际使用的接口名称。
• 替代方案：若线程间资源共享存在问题，可改用multiprocessing模块，通过队列或Manager对象实现进程间通信，逻辑与线程方案一致。

内容的提问来源于stack exchange，提问作者Marcel Leiter