如何用Scapy将捕获的UDP数据包发送至另一台计算机?
解决方案:Scapy捕获UDP包后跨主机传输并写入PCAP
错误原因
你遇到的TypeError: bytes like object is required, not 'Ether'是因为Scapy捕获返回的是封装好的Ether(链路层)数据包对象,而socket的发送方法仅接受原始字节流,直接传递对象必然报错。
方案1:Scapy转字节+Socket传输(推荐)
通过bytes()方法将Scapy数据包对象转为原始字节,再用UDP Socket发送到目标主机,接收端解析字节并写入PCAP。
发送端(捕获并转发)代码
from scapy.all import sniff, UDP import socket # 替换为接收端的局域网IP和自定义端口 TARGET_IP = "192.168.43.xxx" TARGET_PORT = 9000 # 创建UDP Socket sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) def process_packet(pkt): # 仅处理UDP数据包 if UDP in pkt: # 将Scapy数据包转为原始字节 pkt_bytes = bytes(pkt) # 发送到目标主机 sock.sendto(pkt_bytes, (TARGET_IP, TARGET_PORT)) # 启动捕获:指定接口,过滤UDP流量 sniff(iface="eth0", prn=process_packet, filter="udp")
接收端(接收并写入PCAP)代码
from scapy.all import wrpcap, Ether import socket LISTEN_PORT = 9000 OUTPUT_FILE = "captured_traffic.pcap" # 绑定UDP端口 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(("", LISTEN_PORT)) captured_pkts = [] try: print("等待接收数据包...") while True: # 接收最大UDP载荷(65535字节) data, _ = sock.recvfrom(65535) # 将字节解析为Scapy数据包对象 pkt = Ether(data) captured_pkts.append(pkt) # 实时追加写入PCAP文件 wrpcap(OUTPUT_FILE, captured_pkts, append=True) # 可选:清空列表避免内存占用过大 captured_pkts = [] except KeyboardInterrupt: print(f"捕获终止,数据已保存至 {OUTPUT_FILE}")
方案2:直接用Scapy转发数据包
如果两台主机在同一局域网,可直接修改数据包的目的IP,用Scapy内置的send()方法转发(无需Socket):
from scapy.all import sniff, IP, UDP, send TARGET_IP = "192.168.43.xxx" def forward_packet(pkt): if IP in pkt and UDP in pkt: # 修改数据包的目的IP为接收端 pkt[IP].dst = TARGET_IP # 发送IP层数据包(自动处理链路层) send(pkt[IP], verbose=0) sniff(iface="eth0", prn=forward_packet, filter="udp")
接收端可直接用Scapy捕获并写入PCAP:
from scapy.all import sniff, wrpcap OUTPUT_FILE = "received.pcap" def save_packet(pkt): wrpcap(OUTPUT_FILE, pkt, append=True) sniff(iface="eth0", prn=save_packet, filter="udp")
注意事项
- 确保两台主机在手机热点的局域网内IP可达,关闭双方防火墙或开放对应UDP端口
- 捕获时可通过
filter参数缩小范围,比如"udp port 5000"仅捕获指定端口的UDP流量 - 若无需链路层数据,可仅提取IP/UDP层字节(如
bytes(pkt[IP])),减少传输开销
内容的提问来源于stack exchange,提问作者zazmi
相关产品推荐
相关产品推荐

