使用Scapy双网卡转发Profinet流量时丢包断连问题如何解决

你当前使用的bridge_and_sniff是Scapy原生的单线程用户态转发实现，完全无法满足Profinet场景下3000pps的低延迟转发要求，断连重传是必然结果，核心诱因有三点：

• 原生实现采用单线程串行逻辑：抓包、解析、处理、发包全流程在同一个线程执行，前一个包处理完成前不会抓下一个包，高流量下直接占满抓包队列导致丢包
• 未做系统层网卡配置：两块转发网卡默认被内核协议栈接管，会主动产生RST报文、响应不属于本机的ARP/IP请求，同时网卡默认开启的GRO/GSO/校验和offload特性，会导致Scapy抓到的报文是聚合、截断、校验和错误的无效报文，转发后被对端丢弃
• 未做性能参数调优：Scapy默认抓包缓冲区只有几MB，默认轮询间隔1ms，遇到突发流量直接丢包，延迟也达不到Profinet实时通信要求

第一步：系统层前置配置（不做这步改代码无效） #

先执行以下命令配置网卡，关闭内核对转发流量的干预，关闭网卡offload特性：

# 替换为你实际的网卡名
IF1=ens19
IF2=ens20

# 关闭NetworkManager对两块网卡的托管，避免配置被自动篡改
nmcli device set $IF1 managed no
nmcli device set $IF2 managed no

# 清空网卡IP，down掉网卡改配置
ip link set $IF1 down
ip link set $IF2 down
ip addr flush dev $IF1
ip addr flush dev $IF2

# 关闭所有网卡offload特性，避免抓包拿到错误报文
ethtool -K $IF1 rx off tx off gso off gro off tso off rx-checksumming off tx-checksumming off
ethtool -K $IF2 rx off tx off gso off gro off tso off rx-checksumming off tx-checksumming off

# 开启混杂模式，重新启用网卡
ip link set $IF1 promisc on
ip link set $IF2 promisc on
ip link set $IF1 up
ip link set $IF2 up

# 关闭内核反向路由过滤、ICMP重定向，避免内核主动干预转发流量
sysctl -w net.ipv4.conf.$IF1.rp_filter=0
sysctl -w net.ipv4.conf.$IF2.rp_filter=0
sysctl -w net.ipv4.conf.$IF1.accept_redirects=0
sysctl -w net.ipv4.conf.$IF2.accept_redirects=0
sysctl -w net.ipv4.conf.$IF1.send_redirects=0
sysctl -w net.ipv4.conf.$IF2.send_redirects=0

# 丢弃内核主动发送的TCP RST包，避免打断PLC和IO之间的TCP连接
ebtables -A OUTPUT -p IPv4 --ip-proto tcp --tcp-flags RST RST -j DROP

第二步：替换原生单线程转发逻辑 #

用双线程分别处理两个方向的流量，提前初始化发包socket减少重复开销，关闭不必要的协议解析降低延迟，调大抓包缓冲区扛突发流量，优化后代码如下：

from scapy.all import *
import threading
import os

# 固定网卡配置
IF_PLC = "ens19"
IF_IO = "ens20"

# Scapy全局性能调优
conf.sniff_promisc = True
conf.sniff_bufsize = 32 * 1024 * 1024  # 32MB抓包缓冲区，避免队列满丢包
conf.recv_poll_interval = 0.0001  # 抓包轮询间隔设为0.1ms，降低转发延迟
conf.layers.filter([Ether])  # 默认只解析二层头，减少不必要的解析开销

# 提前初始化两个方向的二层发包socket，避免每次发包重复创建
sock_to_plc = conf.L2socket(iface=IF_PLC)
sock_to_io = conf.L2socket(iface=IF_IO)

def pkt_plc_to_io(pkt):
    # 在此处编写PLC到IO方向的报文篡改逻辑
    # 不需要修改的报文直接原包返回，不要加任何阻塞操作
    return pkt

def pkt_io_to_plc(pkt):
    # 在此处编写IO到PLC方向的报文篡改逻辑
    return pkt

def forward_worker(src_iface, dst_sock, xfrm_func):
    # 每个方向独立跑一个抓包转发线程，互不阻塞
    sniff(
        iface=src_iface,
        prn=lambda p: dst_sock.send(xfrm_func(p)),
        store=0,
        count=0,
        nofilter=1  # 关闭内核BPF过滤，降低抓包开销
    )

if __name__ == "__main__":
    if os.geteuid() != 0:
        exit("必须使用root权限运行该脚本")
    
    # 启动两个方向的转发线程
    t_plc2io = threading.Thread(
        target=forward_worker,
        args=(IF_PLC, sock_to_io, pkt_plc_to_io),
        daemon=True
    )
    t_io2plc = threading.Thread(
        target=forward_worker,
        args=(IF_IO, sock_to_plc, pkt_io_to_plc),
        daemon=True
    )
    t_plc2io.start()
    t_io2plc.start()
    t_plc2io.join()
    t_io2plc.join()

注意事项 #

• Profinet实时类报文要求端到端延迟低于10ms，篡改逻辑里绝对不能加阻塞操作，包括同步写日志、打印报文、调用外部接口等。如果需要审计流量，单独开一个异步队列，把需要记录的信息扔到队列里由独立线程处理，不能阻塞转发主路径
• 脚本运行前先确认没有其他程序占用两块网卡，包括tcpdump、其他抓包程序，否则会争抢抓包资源导致丢包
• 如果后续要解析Profinet上层报文，只在篡改逻辑里按需调用pkt.payload解析，不要全局开启所有协议解析，会大幅增加CPU开销

内容的提问来源于stack exchange，提问作者Eric-Hoffmann