如何扩展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
相关产品推荐
相关产品推荐

