如何结合Scapy与Netfilter Queue重组TCP分片（开发IPS）

Netfilter Queue传过来的是单条TCP报文段，内核已经帮你做了IP分片重组，但TCP流的重组得自己在用户态搞定——毕竟Scapy的TCPStream Session是给嗅探场景设计的，直接套到NFQ上不兼容，得换个思路处理。

两种可行实现方式 #

1. 直接复用Scapy的TCPStreamSession #

Scapy的TCPStreamSession本身就是用来维护TCP流状态、重组分段的，你可以手动实例化这个会话，把NFQ过来的每个Scapy包喂进去就行：

from scapy.layers.inet import IP, TCP
from scapy.sessions import TCPStreamSession
from netfilterqueue import NetfilterQueue

# 初始化TCP流会话
stream_session = TCPStreamSession()

def handle_nfq_packet(nfq_pkt):
    # 把NFQ数据包转成Scapy格式
    scapy_pkt = IP(nfq_pkt.get_payload())
    
    # 交给TCPStreamSession处理，自动维护流状态
    stream_session.on_packet_received(scapy_pkt)
    
    # 获取当前包所属的流标识
    stream_id = stream_session.sessionid(scapy_pkt)
    stream = stream_session.streams.get(stream_id)
    
    if stream and stream.buf:
        # 这里可以对重组后的完整流载荷做IPS检测
        if b"malicious_payload" in stream.buf:
            nfq_pkt.drop()
            return
        # 注意：处理完后要按需清空buffer的已处理部分，避免重复检测
        # 比如如果是HTTP请求，可以处理完header+body后清空对应内容
    
    # 没问题就放行
    nfq_pkt.accept()

# 绑定Netfilter Queue队列（队列号1，需要配合iptables规则）
nfq = NetfilterQueue()
nfq.bind(1, handle_nfq_packet)
try:
    nfq.run()
except KeyboardInterrupt:
    nfq.unbind()

要点：stream.buf会缓存流的未处理数据，你得自己控制消费逻辑，不然数据会一直堆积，影响后续检测。

2. 手动维护TCP流状态 #

如果觉得Scapy的会话不够灵活，比如要定制流重组规则，可以自己写逻辑维护每个流的状态：

from scapy.layers.inet import IP, TCP
from netfilterqueue import NetfilterQueue

# 用字典存储每个TCP流的状态，键是(src_ip, src_port, dst_ip, dst_port)
tcp_stream_cache = {}

def handle_nfq_packet(nfq_pkt):
    scapy_pkt = IP(nfq_pkt.get_payload())
    if not scapy_pkt.haslayer(TCP):
        nfq_pkt.accept()
        return
    
    ip = scapy_pkt[IP]
    tcp = scapy_pkt[TCP]
    stream_key = (ip.src, tcp.sport, ip.dst, tcp.dport)
    
    # 初始化流状态（首次收到该流的包）
    if stream_key not in tcp_stream_cache:
        tcp_stream_cache[stream_key] = {
            "expected_seq": tcp.seq + len(tcp.payload),
            "buffer": b""
        }
    
    stream = tcp_stream_cache[stream_key]
    # 处理有序的TCP段
    if tcp.seq == stream["expected_seq"]:
        stream["buffer"] += bytes(tcp.payload)
        stream["expected_seq"] = tcp.seq + len(tcp.payload)
        
        # 示例：按换行分割处理文本类载荷（比如HTTP、SSH命令）
        while b"\n" in stream["buffer"]:
            line, stream["buffer"] = stream["buffer"].split(b"\n", 1)
            if b"unauthorized_command" in line:
                nfq_pkt.drop()
                return
    # 处理乱序段：可以缓存起来，后续收到连续段时再拼接（这里简化处理）
    else:
        pass
    
    nfq_pkt.accept()

# 启动NFQ
nfq = NetfilterQueue()
nfq.bind(1, handle_nfq_packet)
try:
    nfq.run()
except KeyboardInterrupt:
    nfq.unbind()

这种方式需要自己处理TCP的序号、乱序、重传等细节，但灵活性更高，适合需要深度定制流处理逻辑的场景。

关键注意事项 #

• IP分片不用管：Netfilter默认会在内核层面重组IP分片，所以你拿到的都是完整的IP包，不用自己处理IP分片重组。如果非要关闭内核重组，可以改sysctl net.ipv4.ip_defrag=0，但不推荐，会增加复杂度。
• 流超时清理：不管用哪种方式，都要定期清理长时间无活动的TCP流（比如30分钟没数据的），不然内存会越用越多。
• 性能考量：流重组会增加CPU和内存开销，高流量场景下要优化缓存逻辑，比如用LRU淘汰策略管理流缓存。

内容的提问来源于stack exchange，提问作者TECH HINDER