网络协议逆向分析中,关键词字典不是简单的字符串列表。它承担着快速标注字段类型、边界和语义的作用。拿到一段未知TCP载荷,如果字典里已经有Content-Length、USER、PASS、MODE这类协议特征词,分析脚本就能在读取字节流的同时给出候选解释。Ruby在这类任务里有不错的文本处理能力和灵活的集合类,适合先搭建一个轻量但可扩展的字典库。

一、协议关键词字典在逆向场景中的作用
协议逆向时,样本流量往往混杂着二进制数据和可读字符串。直接逐字节观察效率低,而且容易忽略重复出现的字段。关键词字典通过预先收集目标协议中的常量、命令、状态码、参数名,让脚本在读取数据时自动做模式匹配。例如FTP会话中的220、USER、331、PASS,HTTP请求中的GET、Host、Cookie,以及某些私有协议里的固定魔数。字典越准确,逆向分析越能聚焦在真正未知的字段上。
另一方面,关键词字典还可以作为协议格式推断的前置步骤。当某个关键词在多个会话中重复出现在固定偏移处,就很可能对应一个固定头部或命令字段;当它后面紧跟的数据长度变化明显,又可能指向变长字段。把匹配结果与偏移、会话方向、时序放在一起看,能帮助判断字段边界。Ruby的字符串和数组操作适合处理这类字节流切片、搜索和统计。
二、用Ruby构建字典库的核心数据结构
先明确字典库的基本能力:存储词条、去重、按前缀搜索、统计来源。最简单的实现可以把关键词放进Array,但每次匹配都要线性遍历,词条达到几千条时会变慢。更合适的做法是用Hash做O(1)精确查询,用Set做去重,再用Trie树支持前缀联想。Ruby标准库提供Set,但没有内置Trie,需要自己实现一个简单版本。
下面给出一段Trie节点的实现。每个节点持有一个children哈希表和一个结束标记,插入和搜索都按字符逐层下降。这样字典库既能判断某个词是否完整存在,也能在用户输入GET时返回GET以及其他以该前缀开头的词条。
class TrieNode
attr_accessor :children, :leaf
def initialize
@children = {}
@leaf = false
end
end
class ProtocolKeywordTrie
def initialize
@root = TrieNode.new
end
def insert(word)
node = @root
word.each_char do |ch|
node.children[ch] ||= TrieNode.new
node = node.children[ch]
end
node.leaf = true
end
def search(word)
node = @root
word.each_char do |ch|
return false unless node.children.key?(ch)
node = node.children[ch]
end
node.leaf
end
def starts_with(prefix)
node = @root
prefix.each_char do |ch|
return [] unless node.children.key?(ch)
node = node.children[ch]
end
collect_words(node, prefix)
end
private
def collect_words(node, prefix)
words = []
words << prefix if node.leaf
node.children.each do |ch, child|
words.concat(collect_words(child, prefix + ch))
end
words
end
end
在这个Trie基础上,可以增加match_in_stream方法,接收一段字节流并返回所有命中的关键词及位置。相比正则一次只能匹配一类模式,Trie扫描可以同时匹配多个词,而且失败时跳过字符,适合在线分析。
三、协议关键词自动提取与词条扩充
手工整理关键词太慢,而且私有协议文档往往不完整。可以先从样本流量中提取可打印字符串,再用规则筛选出候选关键词。常见的做法是读取报文后,将连续的可打印ASCII字符切出来,过滤掉太短、纯数字或纯随机噪声,再统计频率。出现频率高且长度适中的词条,很可能就是协议中的命令名、状态码或参数名。
Ruby实现一个简单的提取器不复杂。下面这段代码遍历报文字节,遇到可打印字符就累积,遇到不可打印字符就把当前累积的字符串作为候选词输出。之后可用uniq和长度过滤来清理结果。
def extract_ascii_strings(data, min_len = 3)
candidates = []
current = String.new
data.each_byte do |byte|
if byte >= 0x20 && byte <= 0x7E
current << byte.chr
else
candidates << current if current.length >= min_len
current = String.new
end
end
candidates << current if current.length >= min_len
candidates
end
def build_candidate_keywords(packets)
packets.flat_map { |pkt| extract_ascii_strings(pkt) }
.reject { |s| s.match?(/\A\d+\z/) }
.tally
.sort_by { |_word, count| -count }
.first(500)
.map(&:first)
end
这种提取方式会混入主机名、路径、错误信息等噪声,所以还需要人工确认或结合协议上下文过滤。对于已经确定的关键词,可以通过字典库的insert方法批量导入,并记录来源样本,方便日后回溯。
四、原始报文中的关键词命中扫描
字典库建好后,核心用途是对新报文做批量扫描。扫描时不只做整词匹配,还要输出每个关键词出现的偏移,因为偏移对字段定位很关键。可以在Trie中增加扫描逻辑:从每个字节位置开始尝试匹配,命中后记录词条、起始偏移和结束偏移。对于长度较短的命令词,还会出现嵌套命中,例如GET和GETTING,这时保留最长命中即可。
下面给出一个简化的流式扫描方法,返回所有命中的哈希数组,每个哈希包含keyword、offset和length。
def scan_keywords(trie, data)
hits = []
i = 0
while i < data.length
node = trie.instance_variable_get(:@root)
j = i
longest = nil
while j < data.length
ch = data[j]
break unless node.children.key?(ch)
node = node.children[ch]
j += 1
if node.leaf
longest = { keyword: data[i...j], offset: i, length: j - i }
end
end
hits << longest if longest
i += 1
end
hits
end
实际使用中,可以把扫描结果与四元组、会话方向一起记录。比如同一个关键词总出现在某个固定偏移,且前后字节变化小,基本可以判断为协议类型标识;如果关键词后面紧跟的数字在逐渐递增,可能是序列号或长度字段。这种结合字典的启发式判断,虽然不能替代完整的协议状态机分析,但能显著缩小人工排查范围。
五、字典库的工程化与维护建议
字典库不是一次性产物。随着新样本加入,旧词条可能失效,新协议特征会不断出现。需要给每个词条维护元数据,例如来源文件、首次出现时间、命中次数、是否为人工确认。Ruby可以用Hash嵌套实现,也可以引入SQLite存储,这样多轮逆向工程可以在同一份字典上协作。
性能方面,如果报文量很大,单线程扫描可能成为瓶颈。可以通过按会话分片处理、缓存常用关键词的Regexp等方式提速。对于超长报文,还可以先定位明文区域,跳过压缩或加密数据段。整个字典库的设计保持输入输出简单,便于集成到现有分析框架中。

