在python中解析ERF捕获文件

2024-09-28 17:05:15 发布

您现在位置:Python中文网/ 问答频道 /正文

在python中解析ERF(endace)捕获文件的最佳方法是什么?我找到了一个python的libpcap包装器,但我不认为lipcap支持ERF格式。在

谢谢!在


Tags: 文件方法格式libpcaperflipcapendace
2条回答

ERF记录可以包含附加到16字节ERF记录头的可选扩展头。“type”字段的高位表示存在扩展头。我在strix的例子中添加了一个扩展头的测试,以及扩展头本身的解码。注意,如果存在扩展头,以太网帧的测试也需要稍微改变。在

注意:我认为ERF记录可以包含多个扩展头,但我不知道要测试这些头。扩展头结构并没有特别好地记录,我所掌握的唯一记录只包含一个扩展名。在

import struct
import scapy.layers.all as sl

def erf_records( f ):
    """
    Generator which parses ERF records from file-like ``f``
    """
    while True:
        # The ERF header is fixed length 16 bytes
        hdr = f.read( 16 )
        if hdr:
            rec = {}
            # The timestamp is in Intel byte-order
            rec['ts'] = struct.unpack( '<Q', hdr[:8] )[0]
            # The rest is in network byte-order
            rec.update( zip( ('type',  # ERF record type
                              'flags', # Raw flags bit field
                              'rlen',  # Length of entire record
                              'lctr',  # Interstitial loss counter
                              'wlen'), # Length of packet on wire
                             struct.unpack( '>BBHHH', hdr[8:] ) ) )
            rec['iface']  = rec['flags'] & 0x03
            rec['rx_err'] = rec['flags'] & 0x10 != 0

            #- Check if ERF Extension Header present.  
            #  Each Extension Header is 8 bytes.
            if rec['type'] & 0x80:
                ext_hdr = f.read( 8 )
                rec.update( zip( (
                        'ext_hdr_signature',     # 1 byte
                        'ext_hdr_payload_hash',  # 3 bytes
                        'ext_hdr_filter_color',  # 1 bye
                        'ext_hdr_flow_hash'),    # 3 bytes
                        struct.unpack( '>B3sB3s', ext_hdr ) ) )
                #- get remaining payload, less ext_hdr
                rec['pkt'] = f.read( rec['rlen'] - 24 )
            else:
                rec['pkt'] = f.read( rec['rlen'] - 16 )
            if rec['type'] & 0x02:
                # ERF Ethernet has an extra two bytes of pad between ERF header
                # and beginning of MAC header so that IP-layer data are DWORD
                # aligned.  From memory, none of the other types have pad.
                rec['pkt'] = rec['pkt'][2:]
                rec['pkt'] = sl.Ether( rec['pkt'] )
            yield rec
        else:
            return

这是一个简单的ERF记录解析器,它每包返回一个dict(我只是把它组合在一起,所以没有经过很好的测试)。不是所有的标志字段都被解码,但是那些没有被解码的字段并不广泛适用):

注意:

  • ERF记录类型:1=HDLC,2=以太网,3=ATM,4=重新组装的AAL5,5-7个多通道变体,此处不处理额外的报头。在
  • ^如果快照长度太短,{}可以小于wlen+len(header)。在
  • 间隙丢失计数器是指当Dag数据包处理器的输入队列溢出时,在该数据包和之前捕获的数据包之间丢失的数据包数。在
  • 如果你不想用scapy,就把这两行注释掉。在

代码:

import scapy.layers.all as sl

def erf_records( f ):
    """
    Generator which parses ERF records from file-like ``f``
    """
    while True:
        # The ERF header is fixed length 16 bytes
        hdr = f.read( 16 )
        if hdr:
            rec = {}
            # The timestamp is in Intel byte-order
            rec['ts'] = struct.unpack( '<Q', hdr[:8] )[0]
            # The rest is in network byte-order
            rec.update( zip( ('type',  # ERF record type
                              'flags', # Raw flags bit field
                              'rlen',  # Length of entire record
                              'lctr',  # Interstitial loss counter
                              'wlen'), # Length of packet on wire
                             struct.unpack( '>BBHHH', hdr[8:] ) ) )
            rec['iface']  = rec['flags'] & 0x03
            rec['rx_err'] = rec['flags'] & 0x10 != 0
            rec['pkt'] = f.read( rec['rlen'] - 16 )
            if rec['type'] == 2:
                # ERF Ethernet has an extra two bytes of pad between ERF header
                # and beginning of MAC header so that IP-layer data are DWORD
                # aligned.  From memory, none of the other types have pad.
                rec['pkt'] = rec['pkt'][2:]
                rec['pkt'] = sl.Ether( rec['pkt'] )
            yield rec
        else:
            return

相关问题 更多 >