如何让Scapy的sr函数识别自定义协议的响应数据包？

核心问题分析 #

• 你发送的是广播UDP包，IP层默认的校验逻辑会要求响应包的源IP必须等于请求的目的IP（也就是广播地址192.168.1.255），但实际设备回复的包源IP是它自身的IP，这就导致IP层直接判定不是有效响应，自定义层的answers方法根本没机会执行。
• 你只绑定了“UDP目标端口是8888时解析为MyLayer”的规则，而设备响应的UDP包源端口是8888，Scapy不会自动把这部分负载解析成MyLayer，自然也触发不了你的自定义匹配逻辑。

解决方案步骤 #

1. 双向绑定UDP与自定义层 #

在现有绑定代码后添加反向绑定，确保响应包的UDP负载能被解析为MyLayer：

bind_layers(UDP, MyLayer, dport=PORTNUMBER)
# 添加反向绑定，匹配响应包的sport字段
bind_layers(UDP, MyLayer, sport=PORTNUMBER)

2. 自定义响应匹配规则（绕过IP层的严格校验） #

推荐使用sr的lfilter参数自定义匹配逻辑，不用修改全局层的方法，更灵活：
调用sr时，直接指定哪些数据包算有效响应：

results, _ = sr(reqPacket, 
                timeout=2,
                filter=f"udp dst port {PORTNUMBER}", 
                iface="NIC for Test",
                # 自定义过滤：检查UDP层存在、自定义层存在，且通过自定义层的answers校验
                lfilter=lambda pkt: UDP in pkt and MyLayer in pkt and pkt[MyLayer].answers(reqPacket[MyLayer])
)

3. 优化自定义层的answers方法（可选） #

可以让匹配逻辑更严谨，比如校验请求响应的id字段（如果你的协议用id做请求标识）：

def answers(self, other):
    print(">>>> Running answers of MyLayer")
    if isinstance(other, MyLayer):
        # 根据实际协议添加匹配逻辑，比如id一致
        return self.id == other.id
    return 0

完整修改后的示例代码 #

from scapy.all import *

PORTNUMBER = 8888

class MyLayer(Packet):
    name = "MyLayer"
    fields_desc = [BitField("id", 1, 16),
                   FieldLenField("len", None, length_of="data"),
                   StrLenField("data", "", length_from=lambda pkt: pkt.len)]

    def answers(self, other):
        print(">>>> Running answers of MyLayer")
        if isinstance(other, MyLayer):
            # 可根据协议添加id匹配等逻辑
            return self.id == other.id
        return 0

# 双向绑定UDP与MyLayer，确保请求和响应都能解析
bind_layers(UDP, MyLayer, dport=PORTNUMBER)
bind_layers(UDP, MyLayer, sport=PORTNUMBER)

if __name__ == "__main__":
    src = "192.168.1.200"
    dst = "192.168.1.255"

    ipHeader = IP(src=src, dst=dst)
    udpHeader = UDP(sport=PORTNUMBER, dport=PORTNUMBER)
    myLayer = MyLayer(data="abcd1234")
    reqPacket = ipHeader/udpHeader/myLayer

    results, _ = sr(reqPacket, 
                    timeout=2,
                    filter=f"udp dst port {PORTNUMBER}", 
                    iface="NIC for Test",
                    # 自定义过滤逻辑，确保匹配到对应响应
                    lfilter=lambda pkt: UDP in pkt and MyLayer in pkt and pkt[MyLayer].answers(reqPacket[MyLayer])
    )

    # 打印结果验证
    print(f"Received {len(results)} valid responses")
    for req, resp in results:
        print(f"Response: {resp.summary()}")

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