You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何扩展dpkt模块中IEEE80211.Action的action_parser内部方法

问题描述

我正在使用dpkt模块解码pcapng数据包,但该模块部分功能不完善且扩展性较差。当前IEEE80211类内部Action子类的unpack方法中,action_parser仅支持block ack相关的类别与代码,其他情况会抛出异常,导致无法完成部分解码(因该逻辑位于IEEE80211构造流程中)。我希望扩展action_parser以支持更多选项,请问是否有方法从外部修改action_parser的值?或者有其他扩展该代码的方式?虽然后续可向主仓库提交修改,但我暂不希望走此路径。

相关代码
class IEEE80211(dpkt.Packet):

    <snip>

    def unpack(self, buf):
        dpkt.Packet.unpack(self, buf)
        self.data = buf[self.__hdr_len__:]

        m_decoder = {
            M_BEACON: ('beacon', self.Beacon),
            M_ASSOC_REQ: ('assoc_req', self.Assoc_Req),
            M_ASSOC_RESP: ('assoc_resp', self.Assoc_Resp),
            M_DISASSOC: ('diassoc', self.Disassoc),
            M_REASSOC_REQ: ('reassoc_req', self.Reassoc_Req),
            M_REASSOC_RESP: ('reassoc_resp', self.Assoc_Resp),
            M_AUTH: ('auth', self.Auth),
            M_PROBE_RESP: ('probe_resp', self.Beacon),
            M_DEAUTH: ('deauth', self.Deauth),
            M_ACTION: ('action', self.Action)
        }

        <snip>

        decoder = {
            MGMT_TYPE: m_decoder,
            CTL_TYPE: c_decoder,
            DATA_TYPE: d_decoder
        }

        <snip>
        try:
            parser = decoder[self.type][self.subtype][1]
            name = decoder[self.type][self.subtype][0]
        except KeyError:
            raise dpkt.UnpackError("KeyError: type=%s subtype=%s" % (self.type, self.subtype))

    class Action(dpkt.Packet):
        __hdr__ = (
            ('category', 'B', 0),
            ('code', 'B', 0),
        )

        def unpack(self, buf):
            dpkt.Packet.unpack(self, buf)

            action_parser = {
                BLOCK_ACK: {
                    BLOCK_ACK_CODE_REQUEST: ('block_ack_request', IEEE80211.BlockAckActionRequest),
                    BLOCK_ACK_CODE_RESPONSE: ('block_ack_response', IEEE80211.BlockAckActionResponse),
                    BLOCK_ACK_CODE_DELBA: ('block_ack_delba', IEEE80211.BlockAckActionDelba),
                },
            }

            try:
                decoder = action_parser[self.category][self.code][1]
                field_name = action_parser[self.category][self.code][0]
            except KeyError:
                raise dpkt.UnpackError("KeyError: category=%s code=%s" % (self.category, self.code))
解决方案

方法1:猴子补丁扩展action_parser

直接替换Action类的unpack方法,新增自定义解析器的扩展入口,无需修改源码:

import dpkt

# 定义你需要支持的新Action解析类,示例为QoS Action
class QoSAction(dpkt.Packet):
    __hdr__ = (
        ('qos_field', 'H', 0),
        # 根据实际协议字段补充
    )

# 给Action类添加自定义解析器的存储属性
dpkt.ieee80211.IEEE80211.Action.custom_parsers = {}

# 保存原始unpack方法,重写新逻辑
original_unpack = dpkt.ieee80211.IEEE80211.Action.unpack
def extended_unpack(self, buf):
    original_unpack(self, buf)
    
    # 合并默认解析规则与自定义规则
    action_parser = {
        dpkt.ieee80211.BLOCK_ACK: {
            dpkt.ieee80211.BLOCK_ACK_CODE_REQUEST: ('block_ack_request', dpkt.ieee80211.IEEE80211.BlockAckActionRequest),
            dpkt.ieee80211.BLOCK_ACK_CODE_RESPONSE: ('block_ack_response', dpkt.ieee80211.IEEE80211.BlockAckActionResponse),
            dpkt.ieee80211.BLOCK_ACK_CODE_DELBA: ('block_ack_delba', dpkt.ieee80211.IEEE80211.BlockAckActionDelba),
        }
    }
    # 合并自定义解析规则
    for category, codes in self.custom_parsers.items():
        action_parser.setdefault(category, {}).update(codes)
    
    try:
        decoder = action_parser[self.category][self.code][1]
        field_name = action_parser[self.category][self.code][0]
        setattr(self, field_name, decoder(self.data))
        self.data = b''
    except KeyError:
        # 可选:不抛出异常,保留原始data供后续自行解析
        pass

# 替换原unpack方法
dpkt.ieee80211.IEEE80211.Action.unpack = extended_unpack

# 外部添加自定义解析规则,示例:QoS类别(假设值为1)、代码0
dpkt.ieee80211.IEEE80211.Action.custom_parsers[1] = {
    0: ('qos_action', QoSAction)
}

方法2:继承重写Action类并替换

创建自定义Action子类,重写unpack逻辑后替换原类:

import dpkt

# 定义自定义解析类
class QoSAction(dpkt.Packet):
    __hdr__ = (
        ('qos_field', 'H', 0),
    )

class CustomAction(dpkt.ieee80211.IEEE80211.Action):
    def unpack(self, buf):
        dpkt.Packet.unpack(self, buf)
        
        # 扩展后的解析规则
        action_parser = {
            dpkt.ieee80211.BLOCK_ACK: {
                dpkt.ieee80211.BLOCK_ACK_CODE_REQUEST: ('block_ack_request', dpkt.ieee80211.IEEE80211.BlockAckActionRequest),
                dpkt.ieee80211.BLOCK_ACK_CODE_RESPONSE: ('block_ack_response', dpkt.ieee80211.IEEE80211.BlockAckActionResponse),
                dpkt.ieee80211.BLOCK_ACK_CODE_DELBA: ('block_ack_delba', dpkt.ieee80211.IEEE80211.BlockAckActionDelba),
            },
            # 添加自定义类别与解析器
            1: {
                0: ('qos_action', QoSAction)
            }
        }
        
        try:
            decoder = action_parser[self.category][self.code][1]
            field_name = action_parser[self.category][self.code][0]
            setattr(self, field_name, decoder(self.data))
            self.data = b''
        except KeyError:
            # 可选:跳过异常,保留原始数据
            pass

# 替换原Action类
dpkt.ieee80211.IEEE80211.Action = CustomAction

方法3:捕获异常跳过未支持的Action

如果不想修改原类,可在解码时捕获异常,自行处理未支持的Action:

import dpkt
from dpkt import UnpackError

with open('target.pcapng', 'rb') as f:
    pcap_reader = dpkt.pcapng.Reader(f)
    for ts, buf in pcap_reader:
        try:
            eth_frame = dpkt.ethernet.Ethernet(buf)
            if isinstance(eth_frame.data, dpkt.ieee80211.IEEE80211):
                # 正常解码逻辑
                pass
        except UnpackError as e:
            if "category=" in str(e):
                # 处理未支持的Action,比如提取原始数据手动解析
                print(f"跳过未支持的Action: {e}")
            else:
                # 其他解码错误,按需抛出或处理
                raise

内容的提问来源于stack exchange,提问作者quite68

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.18 03:10:34