使用Tshark和Python实现网络数据包十六进制字节与协议层数据的精细映射

tshark -r input.pcap -T pdml > output.pdml

import xml.etree.ElementTree as ET
import subprocess
import os

def convert_pcap_to_pdml(pcap_file, pdml_file):
    """
    使用tshark将pcap文件转换为pdml文件。
    :param pcap_file: 输入的PCAP文件路径。
    :param pdml_file: 输出的PDML文件路径。
    """
    if not os.path.exists(pcap_file):
        print(f"错误: PCAP文件 '{pcap_file}' 不存在。")
        return False

    try:
        # 使用subprocess运行tshark命令
        # stdout参数直接将tshark的输出写入pdml_file
        with open(pdml_file, 'w', encoding='utf-8') as f_out:
            subprocess.run(['tshark', '-r', pcap_file, '-T', 'pdml'], 
                           stdout=f_out, 
                           check=True,
                           encoding='utf-8') # 确保输出编码正确
        print(f"成功将 '{pcap_file}' 转换为 '{pdml_file}'")
        return True
    except subprocess.CalledProcessError as e:
        print(f"Tshark转换失败，命令返回非零退出码: {e}")
        print(f"Stderr: {e.stderr.decode()}") # 打印tshark的错误输出
        return False
    except FileNotFoundError:
        print("错误: 'tshark' 命令未找到。请确保tshark已安装并配置到系统PATH中。")
        return False
    except Exception as e:
        print(f"发生未知错误: {e}")
        return False

def parse_pdml_for_field_info(pdml_file):
    """
    解析PDML文件，提取每个数据包中每个字段的名称、显示值、起始位置和长度。
    :param pdml_file: 输入的PDML文件路径。
    :return: 包含所有数据包字段信息的列表，每个元素代表一个数据包的字段列表。
    """
    if not os.path.exists(pdml_file):
        print(f"错误: PDML文件 '{pdml_file}' 不存在。")
        return []

    packets_info = []
    try:
        tree = ET.parse(pdml_file)
        root = tree.getroot()

        for packet_elem in root.findall('packet'):
            packet_fields = []
            for proto_elem in packet_elem.findall('proto'):
                for field_elem in proto_elem.findall('field'):
                    pos_str = field_elem.get('pos')
                    size_str = field_elem.get('size')

                    if pos_str is not None and size_str is not None:
                        try:
                            pos = int(pos_str)
                            size = int(size_str)
                            field_info = {
                                'name': field_elem.get('name'),
                                'showname': field_elem.get('showname', field_elem.get('name')), # 优先使用showname
                                'value': field_elem.get('value'),
                                'display_value': field_elem.get('show'),
                                'pos': pos,
                                'size': size,
                                'layer_name': proto_elem.get('name') # 字段所属协议层
                            }
                            packet_fields.append(field_info)
                        except ValueError:
                            # 忽略pos或size不是有效数字的字段
                            continue
            packets_info.append(packet_fields)
    except ET.ParseError as e:
        print(f"解析PDML文件时发生XML错误: {e}")
        return []
    except Exception as e:
        print(f"解析PDML文件时发生未知错误: {e}")
        return []

    return packets_info

# 示例使用
if __name__ == "__main__":
    pcap_path = "sample.pcap" # 替换为你的PCAP文件路径
    pdml_path = "sample.pdml"

    # 1. 模拟创建一个空的pcap文件以供测试 (实际使用时请替换为真实的pcap文件)
    # 或者确保你有一个真实的pcap文件在这里
    if not os.path.exists(pcap_path):
        print(f"创建模拟PCAP文件: {pcap_path}")
        # 这里只是一个占位符，tshark需要一个有效的pcap文件
        # 如果没有真实的pcap文件，tshark转换会失败
        # 实际操作中，请确保 'sample.pcap' 是一个有效的网络抓包文件
        with open(pcap_path, 'wb') as f:
            f.write(b'\xd4\xc3\xb2\xa1\x01\x02\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00') # pcap global header
            # 这是一个非常简化的，可能不被tshark完全识别的pcap文件头
            # 更好的做法是使用scapy创建一个简单的pcap文件
            # from scapy.all import Ether, IP, TCP, wrpcap
            # packets = [Ether()/IP()/TCP()]
            # wrpcap(pcap_path, packets)


    # 2. 转换PCAP到PDML
    if convert_pcap_to_pdml(pcap_path, pdml_path):
        # 3. 解析PDML并获取字段信息
        all_packets_field_info = parse_pdml_for_field_info(pdml_path)

        if all_packets_field_info:
            print(f"\n成功解析 {len(all_packets_field_info)} 个数据包的字段信息。")
            # 打印第一个数据包的部分字段信息作为示例
            if all_packets_field_info[0]:
                print("\n第一个数据包的字段信息示例:")
                for i, field in enumerate(all_packets_field_info[0][:5]): # 只打印前5个字段
                    print