在分析网络问题时，Wireshark的Follow TCP Stream功能几乎人人都在用，它能把一次TCP会话的双向数据完整还原出来。但如果要在生产环境中做自动化的流量审计、协议分析或者入侵检测，就不能依赖图形界面工具，需要自己实现抓包和流重组的逻辑。Ruby虽然不是网络编程领域最常见的语言，但借助libpcap提供的底层能力，完全可以实现一套轻量级的TCP会话分析工具。本文将从抓包原理、数据包解析、流重组算法三个层面逐步展开，并给出完整的示例代码。
一、Ruby环境下抓包方案的选择与环境搭建
在Ruby生态中做抓包，主要有三条路可走。第一条是直接调用系统命令tcpdump，把结果落成pcap文件再离线解析，这种方式实现最简单，但实时性差，且无法精细控制过滤规则。第二条是使用PCAPRUB这个gem，它是对libpcap的C语言绑定，性能好，能实时捕获数据包，也是本文推荐的方式。第三条是纯Ruby实现的packetfu库，它更擅长对数据包进行构造和解析，抓包能力则依赖前两者。
以PCAPRUB为例，安装非常直接，执行gem install pcaprub即可。需要注意的一点是，编译过程需要系统已经安装libpcap的开发头文件，Debian或Ubuntu下可以通过sudo apt-get install libpcap-dev安装，CentOS下对应的包名是libpcap-devel。如果在macOS上开发，系统自带的libpcap通常可以满足需求，但建议通过Homebrew安装新版本以获得对较新内核特性的支持。
抓包本身需要一定的系统权限。Linux上普通用户没有直接访问网卡的权限，要么以root身份运行脚本，要么通过setcap给Ruby解释器赋予抓包能力，例如执行sudo setcap cap_net_raw,cap_net_admin=eip $(which ruby)，后者在安全性上更可控，避免了整个脚本以root权限运行。
二、逐层解析数据包：从以太网帧到TCP头
libpcap交给应用层的是一段原始字节，要还原TCP会话，第一步是把各个协议层的头部剥开。以太网帧头固定14字节，前6字节是目的MAC地址，接着6字节是源MAC地址，最后2字节是上层协议类型。协议类型字段值为0x0800表示上层是IPv4，这是我们关心的情形；如果是0x8100则说明外面还套了一层VLAN标签，需要多跳过4字节再解析。
IPv4头部的第一个字节高4位是版本号，低4位是头长度（以4字节为单位），通过它可以计算出IP头实际占多少字节，这个长度是可变的，因为IP头允许携带选项字段。IP头中还有两个关键字段：8位协议号，值为6表示TCP；以及16位总长度字段，用总长度减去头长度就得到TCP段的字节数。TCP头部的标准长度是20字节，它的第12字节高4位同样是以4字节为单位的头长度，数据偏移就由此计算得出，头部之后的全部内容就是应用层载荷。
Ruby提供了String#unpack方法来按格式解析二进制数据，非常契合这个场景。下面的代码演示了完整的解析过程：
require 'pcaprub'
def parse_packet(pkt)
# 解析以太网帧头，14字节
eth_type = pkt[12, 2].unpack('n').first
offset = 14
return nil unless eth_type == 0x0800 # 只处理IPv4
ip = pkt[offset..]
version_ihl = ip[0].unpack1('C')
ihl = (version_ihl & 0x0f) * 4 # IP头长度
total_len = ip[2, 2].unpack1('n') # IP总长度
proto = ip[9].unpack1('C')
src_ip = ip[12, 4].unpack('C4').join('.')
dst_ip = ip[16, 4].unpack('C4').join('.')
return nil unless proto == 6 # 只处理TCP
tcp = ip[ihl..]
src_port, dst_port = tcp[0, 4].unpack('nn')
data_offset = (tcp[12].unpack1('C') >> 4) * 4 # TCP头长度
seq = tcp[4, 4].unpack1('N') # 序列号
ack = tcp[8, 4].unpack1('N') # 确认号
flags = tcp[13].unpack1('C')
payload = tcp[data_offset.., total_len - ihl - data_offset]
{ src_ip: src_ip, dst_ip: dst_ip, src_port: src_port, dst_port: dst_port,
seq: seq, ack: ack, flags: flags, payload: payload }
end解析过程中有个容易踩的坑：不要直接用捕获数据包的总长度来切割payload，以太网帧有最小长度要求，短包会被填充，导致尾部多出无意义的字节。正确的做法是始终以IP头中的总长度字段为准，上面的代码已经体现了这一点。
三、TCP流重组的核心算法与实现
有了逐包的解析结果，接下来才是本文的核心——流重组。TCP连接由四元组（源IP、源端口、目的IP、目的端口）唯一标识，但由于抓包视角下两个方向的数据都会出现，实践中通常把四元组规范化：将两端地址按字典序排序后拼接作为会话key，这样同一连接的双向报文就会落入同一个会话桶中。再配合一个方向标记，就能分别保存客户端到服务端、服务端到客户端两条单向数据流。
重组算法的关键在于处理乱序、重传和重复这三种情况。核心思路是维护每个方向上的"期望序列号"：收到报文时，如果其序列号等于期望值，说明数据连续，直接追加到缓冲区，期望值前移报文载荷长度；如果序列号小于期望值，说明是重传或部分重复，只截取尚未接收的尾部；如果序列号大于期望值，说明中间出现了空洞，需要把报文暂存到一个按序列号排序的乱序队列里，等缺失的报文到达后再统一补齐。SYN和FIN标志会消耗一个序列号，处理时不能遗漏。
下面是一个简化但可运行的重组器实现，配合PCAPRUB的实时捕获循环使用：
class TcpReassembler
def initialize
@sessions = Hash.new { |h, k| h[k] = { data: '', expected: nil, queue: [] } }
end
def session_key(pkt)
a = "#{pkt[:src_ip]}:#{pkt[:src_port]}"
b = "#{pkt[:dst_ip]}:#{pkt[:dst_port]}"
[a, b].sort.join('-')
end
def feed(pkt)
key = session_key(pkt)
stream = @sessions[key]
seq, payload = pkt[:seq], pkt[:payload]
if (pkt[:flags] & 0x02) != 0 # SYN包，初始化期望序列号
stream[:expected] = seq + 1
return
end
return if stream[:expected].nil? || payload.nil? || payload.empty?
if seq == stream[:expected]
append(stream, seq, payload)
drain_queue(stream)
elsif seq + payload.bytesize > stream[:expected]
# 部分重叠，截取新增部分
overlap = stream[:expected] - seq
append(stream, stream[:expected], payload[overlap..])
drain_queue(stream)
else
stream[:queue] << [seq, payload] # 完全重复，忽略即可
end
@sessions[key]
end
private
def append(stream, seq, payload)
stream[:data] << payload
stream[:expected] = seq + payload.bytesize
end
def drain_queue(stream)
until stream[:queue].empty?
seq, payload = stream[:queue].min_by { |s, _| s }
break if seq > stream[:expected]
stream[:queue].delete([seq, payload])
next if seq + payload.bytesize <= stream[:expected]
overlap = stream[:expected] - seq
append(stream, stream[:expected], payload[[overlap, 0].max..])
end
end
end
cap = PCAPRUB::Pcap.open_live('eth0', 65535, true, 1000)
cap.setfilter('tcp port 80')
reassembler = TcpReassembler.new
cap.each_packet do |raw|
pkt = parse_packet(raw.raw_data)
next if pkt.nil?
stream = reassembler.feed(pkt)
end四、会话分析实践与性能优化建议
流重组完成后，会话分析就是在此基础上做上层逻辑了。典型应用包括：统计会话持续时间和双向流量大小；基于HTTP请求行的特征识别应用层协议；对payload做关键字扫描实现简单的敏感信息审计。还可以结合时间戳做行为分析，例如检测端口扫描——短时间内同一源地址与大量不同端口建立连接的行为在会话表中会表现得非常明显。
在性能方面，Ruby的字符串操作效率不如C，处理高流量场景时需要一些取舍。首先是过滤器要尽量下推到内核层，通过setfilter设置BPF过滤表达式，让libpcap在内核中丢弃无关包，只把感兴趣的包交给Ruby层。其次是会话表要有老化机制，为每个会话记录最后活跃时间，定期清理超时会话，避免内存无限增长。如果吞吐量仍是瓶颈，可以考虑只解析头部、按需解析payload，或者把采集层换成Go、Rust实现，Ruby层只负责分析逻辑。
最后要提醒的是，抓包涉及隐私和合规问题。在生产环境部署任何流量分析工具前，务必确认符合所在组织的网络安全策略和相关法律法规，尽量缩小捕获范围，只采集完成分析所必需的数据。通过本文的方案，你已经可以搭建起一个从原始数据包到完整TCP会话的分析管道，后续无论是深入协议细节还是构建检测规则，都有了扎实的基础。