做过网络协议分析或者流量审计的开发者,大概率都跟PCAP文件打过交道。Wireshark抓下来的包、tcpdump落盘的数据,本质上都是一个PCAP格式的二进制文件。但PCAP毕竟是二进制格式,想直接用脚本分析并不方便,很多时候我们需要把它转成CSV、JSON,或者升级成更现代的PCAPNG格式。本文就用Ruby从零实现一个PCAP格式转换工具,把文件头解析、数据包遍历、格式输出这条链路完整走一遍。

一、先搞清楚PCAP文件的二进制结构
写解析器之前必须先理解格式本身。PCAP文件由一个24字节的全局文件头(Global Header)加上一系列数据包记录组成。全局头包含魔数、主版本号、次版本号、时区偏移、时间戳精度、快照长度和网络链路类型这几个字段。其中魔数是判断字节序的关键:如果读到的是十六进制0xD4C3B2F1,说明文件是小端序存储;如果是0xA1B2C3D4,则是大端序。很多新手解析PCAP失败,十有八九是没处理字节序问题。
每个数据包记录由16字节的包头和变长的包数据组成。包头包含四个字段:时间戳秒数、时间戳微秒数、捕获长度(实际抓到的字节数)和原始长度(线上真实字节数)。捕获长度可能小于原始长度,这是因为抓包时设置了snaplen截断,解析时必须以后者为准判断包是否被截断。
链路类型字段(Link Type)决定了包数据的起始协议层,比如1代表以太网帧,101代表原始IP,113代表Linux SLL。如果转换工具要解析到IP层甚至TCP层,就必须根据链路类型做不同的偏移处理,这一点后面实现时会具体展开。
二、用Ruby解析PCAP文件头与数据包
Ruby标准库里的String#unpack和IO#read配合起来处理二进制文件非常顺手。先用unpack解析全局头,判断字节序后再决定后续字段用v还是n模板去解。下面是文件头解析的核心代码:
class PcapReader
MAGIC_LITTLE = 0xD4C3B2F1
MAGIC_BIG = 0xA1B2C3D4
attr_reader :link_type, :snaplen
def initialize(path)
@file = File.open(path, 'rb')
parse_global_header
end
def parse_global_header
raw = @file.read(24)
raise 'not a pcap file' if raw.nil? || raw.bytesize != 24
magic = raw[0, 4].unpack('N').first
case magic
when MAGIC_LITTLE then @endian = 'vV'
when MAGIC_BIG then @endian = 'nN'
else raise "unsupported magic: 0x#{magic.to_s(16)}"
end
vmaj, vmin, _tz, _sig, @snaplen, @link_type =
raw[4, 20].unpack("#{@endian[0]}#{@endian[0]}#{"a4"}a4#{@endian[1]}#{@endian[1]}")
# 简化写法:分别解包每个字段更清晰
@thiszone, @sigfigs = 0, 0
end
def each_packet
while header = @file.read(16)
ts_sec, ts_usec, incl_len, orig_len = header.unpack("#{@endian[1]}#{@endian[1]}#{@endian[1]}#{@endian[1]}")
data = @file.read(incl_len)
break if data.nil? || data.bytesize < incl_len
yield ts_sec, ts_usec, incl_len, orig_len, data
end
end
def close
@file.close
end
end
上面这段代码用逐包读取的方式实现了流式遍历,内存中永远只保留一个包的数据,即使处理几个GB的抓包文件也不会把内存撑爆。这是处理大文件的关键设计。时间戳方面,PCAP标准里微秒字段在有纳秒精度的变体中会被解释为纳秒,判断依据是魔数变成0xA1B23C4D,如果对时间精度有要求,需要额外支持这个变体。
Ruby社区也有现成的抓包库packetfu,安装命令是gem install packetfu。它封装了从链路层到应用层的协议解析,可以直接拿以太网帧解析出IP头和TCP头。不过packetfu偏重于包构造和注入,读取大PCAP文件时性能一般,如果是简单的格式转换,自己解析文件头反而更快更轻量。两者可以结合使用:自研解析器负责遍历,packetfu负责深度协议解码。
三、实现CSV与JSON格式输出
解析出数据包之后,输出的设计就灵活了。CSV适合导入Excel或者数据库做统计,JSON适合喂给其他脚本或日志系统。对于CSV输出,建议至少包含序号、时间戳、包长度、协议类型和源目的地址这几列。下面是转换主逻辑:
require 'csv'
require 'json'
class PcapConverter
def initialize(reader)
@reader = reader
end
def to_csv(out_path)
CSV.open(out_path, 'w') do |csv|
csv << ['index', 'timestamp', 'cap_len', 'orig_len', 'src_ip', 'dst_ip', 'protocol']
index = 0
@reader.each_packet do |ts_sec, ts_usec, incl, orig, data|
index += 1
src, dst, proto = extract_ip_info(data)
csv << [index, format('%d.%06d', ts_sec, ts_usec), incl, orig, src, dst, proto]
end
end
end
def to_json(out_path)
packets = []
@reader.each_packet do |ts_sec, ts_usec, incl, orig, data|
src, dst, proto = extract_ip_info(data)
packets << {
timestamp: format('%d.%06d', ts_sec, ts_usec),
cap_len: incl, orig_len: orig,
src_ip: src, dst_ip: dst, protocol: proto
}
end
File.write(out_path, JSON.pretty_generate(packets))
end
private
# 仅处理以太网链路类型,IP头从偏移14开始
def extract_ip_info(data)
return ['', '', ''] unless @reader.link_type == 1 && data.bytesize > 34
ip = data[14..-1]
version_ihl = ip[0].ord
return ['', '', ''] unless version_ihl >> 4 == 4
total_len = ip[2, 2].unpack('n').first
proto = ip[9].ord
src = ip[12, 4].unpack('C4').join('.')
dst = ip[16, 4].unpack('C4').join('.')
[src, dst, proto == 6 ? 'TCP' : proto == 17 ? 'UDP' : "IP-#{proto}"]
rescue
['', '', '']
end
end
这里有个细节值得注意:JSON输出一次性把所有包塞进数组,大文件会占用大量内存。改进方案是手写流式JSON,即每处理一个包就写入一行,各包之间用换行分隔,也就是常说的JSON Lines格式。这样下游脚本可以逐行读取,两边都不用把全量数据加载进内存,工程上更实用。
另一个容易忽略的点是截断包的处理。当incl_len小于orig_len时,说明包数据不完整,IP头之后的传输层可能缺失。上面的代码在提取地址前做了长度检查,但更严谨的做法是在输出里加一个truncated标记字段,让下游分析工具能够识别并过滤掉这些残缺包,避免统计结果出现偏差。
四、命令行封装与性能优化建议
工具最终要给人用,套一个命令行入口会更友好。Ruby自带的optparse就能满足需求,支持输入文件、输出格式、过滤协议等参数。示例用法设计成ruby pcap_tool.rb input.pcap -f json -o output.json这种形式,代码如下:
require 'optparse'
options = { format: 'csv' }
OptionParser.new do |opts|
opts.banner = 'Usage: ruby pcap_tool.rb input.pcap [options]'
opts.on('-f FORMAT', %w[csv json jsonl], 'output format: csv, json, jsonl') { |v| options[:format] = v }
opts.on('-o FILE', 'output file path') { |v| options[:out] = v }
opts.on('-p PROTO', 'filter protocol: tcp or udp') { |v| options[:proto] = v }
end.parse!
input = ARGV[0] or abort 'missing input file'
reader = PcapReader.new(input)
converter = PcapConverter.new(reader)
case options[:format]
when 'csv' then converter.to_csv(options[:out] || 'output.csv')
when 'json' then converter.to_json(options[:out] || 'output.json')
end
reader.close
性能方面,如果是千万级包量的文件,逐包调用unpack的开销会显现出来。可以考虑两个优化方向:一是用String#unpack1替代返回单元素的unpack,减少数组分配;二是把包头四个字段合并成一次解包调用,模板写成'VVVV',一次系统调用少分配三个中间对象。追求极致性能的话还可以引入nio4r做零拷贝读取,但对于大多数场景,标准库的速度已经足够。
最后提一下扩展方向。有了这套解析框架,往上加功能都很顺:支持PCAPNG输出只需按新格式写文件头和增强版块结构;支持BPF过滤可以移植rbpcap相关的思路;想做协议统计只要在遍历回调里累加计数器即可。理解了PCAP的二进制本质之后,格式转换只是最基础的应用,真正有价值的是基于这套数据做的自动化分析流水线,比如异常流量检测、延迟统计和会话重建,这些都可以在这个工具的骨架上继续生长。