镇海区建设交通局网站企业信息公示网查询官网

​ 作者：高玉涵

​ 时间：2023.9.17 10:25

​ 环境：Linux kali 5.15.0-kali3-amd64，Python 3.11.4，scapy 2.5.0

只有在拿到一些数据之后事情才会变得有趣起来。

前言

​ 通常我在网络嗅探与数据包分析中，使用 Wireshark 就可以很方便地浏览 pcap 文件的内容。但当捕获得流量很大或数据包特征不太明显，再或者数据包特征已确定，要从中进一步分析（提取）流量。以往采用人工方式可以说是种恶梦。幸运的是 Philippe Biondi 为 Python 开发的数据包处理库 Scapy 以精巧和令人惊叹，一两行代码就能解决上述问题（功能远远不止如此）。这里我会演示如何借助 Scapy 的 pcap 数据处理能力，从嗅探到的 HTTP 流量中提取图片。

​ 建议你在 Linux 上使用 Scapy，因为它最初就是按照兼容 Linux 来设计的。最新版本的 Scapy 虽然支持 Windows，但是我假设你使用的和我一样的环境。

一、网络环境示例

图 1-1 网络架构

二、嗅探流量示例

​ 为了感受一下 Scapy 的用法，我先给出用 Wireshark 嗅探的流量，图 2-1 是原始流量样式，图 2-2 为指定一个数据包过滤器（这里是 HTTP）表示要进一步分析的流量。

图 2-1 原始流量样式

图 2-2 HTTP 流量样式

三、pcap 文件处理

​ 我会试着从 HTTP 流量中提取出图片文件，为此编写分析 pcap 文件的 recapper.py 程序，定位数据流中出现的所有图片，并把它们保存到磁盘上。在接下来的代码中，我将用到 namedtuple（命名元组），它是 Python 的一种数据结构，可以通过名称来访问某个字段。标准的元组（tuple）是用来存储一串不可变的值的，跟列表（list）差不多，只是没法修改里面的数据。使用标准元组时，要用数字索引来访问其内部的字段：

point = (1.1, 2.5)
print(point[0], point[1])

​ 要命名元组跟标准元组基本相同，唯一的区别是可以通过属性名来访问它的字段。它能让你写出可读性更高的代码，却又比字典（dictionary）

消耗的内存更少。创建命名元组需要两个参数：元组本身的名字，以及由逗号分隔的若干字段名。举个例子，假设你想定义一个名为 Point 的数据结构，它有两个属性：x 和 y。你可以这样定义:

Point = namedtuple('Point', ['x', 'y'])

​ 接着你可以创建一个，比方说叫 p 的 Point 命名元组，p = Point(35, 65)，然后使用 p.x 和 p.y 来访问它的 x 和 y 属性，就像是在访问一个类的属性一样。这比使用一堆数字索引来访问标准元组要好懂得多。在接下来的示例代码中，我们将创建一个名为 Response 的命名元组：

Response = namedtuple('Response', ['header', 'payload'])

​ 现在，无须用数字索引，直接用 Response.header 和 Response.payload 就能访问 Response 的成员数据，大大改善了代码的可读性。

​ 接下来是 recapper.py 文件，代码如下：

from scapy.all import TCP, rdpcap
import collections
import os
import re
import sys
import zlibOUTDIR = '/home/kali/dev/python/scapy/pic'
PCAPS = '/home/kali/dev/python/scapy'Response = collections.namedtuple('Response', ['header', 'payload'])def get_header(payload):passdef extract_content(Response, content_name='image'):passclass Recapper:def __init__(self, fname):passdef get_responses(self):passdef write(self, content_name):passif __name__ == '__main__':pfile = os.path.join(PCAPS, 'face.pcap')recapper = Recapper(pfile)recapper.get_responses()recapper.write('image')

​ 这是整个脚本的主要框架，之后我会往里面填充辅助函数。首先，导入所需的库，然后指定保存图片的目录和 pcap 文件的路径。接着，定义一个名为 Response 的命名元组，它有两个属性：数据包的头（header）和载荷（payload）。我编写两个辅助函数，分别负责获取数据包头和提取数据包内容。这两个函数将用在 Recapper 类里，而这个类负责重构在数据包流中出现过的图片。除了 __init__ 函数外，Recapper 类还有两个函数：get_responses，负责从 pcap 文件中读取响应数据；write，负责把在响应数据中找到的图片写到输出目录里。

​ 现在我开始写 get_header函数：

def get_header(payload):try:header_raw = payload[:payload.index(b'\r\n\r\n')+2]# 切片前包后不包,+2 是为了将分隔数据行的'\r\n'也含进去except ValueError:sys.stdout.write('-')sys.stdout.flush()return Noneheader = dict(re.findall(r'(?P<name>.*?):(?P<value>.*?)\r\n', header_raw.decode()))if 'Content-Type' not in header:return Nonereturn header

​ get_header函数会读取原始 HTTP 流量，并把 HTTP 头数所单独切出来。我提取 HTTP 头的方式是，从数据包开头一路往下找两个连续的 “\r\n”，把这整段数据切出来。图 3-1 HTTP 报文结构。图 3-2 真实示例。

图 3-1 HTTP 报文结构

图 3-2 真实示例

​ 如果拿到的数据里不存在两个连续的 ”\r\n“，就会产生一个 ValueError 异常，这时我只会在屏幕上输出一个横杠（-），然后返回。如果没有发生异常，我就创建一个名为 header 的字典，把 HTTP 头里的每一行以冒号分割，冒号左边的是字段名，右边的是字段值，按这样的方式存进 header 字典里。如果 HTTP 头里面没有名为 Content-Type 的字段（图 3-3 Content-Type 字段），就返回 None，表示数据包里没有我感兴趣的内容。现在写一个函数，从响应数据里提取内容：

def extract_content(Response, content_name='image'):content, content_type = None, Noneif content_name in Response.header['Content-Type']:content_type = Response.header['Content-Type'].split('/')[1]content = Response.payload[Response.payload.index(b'\r\n\r\n')+4:]if 'Content-Encoding' in Response.header:if Response.header['Content-Encoding'] == 'gzip':content = zlib.decompress(Response.payload, zlib.MAX_WBITS | 32)elif Response.header['Content-Encoding'] == 'deflate':content = zlib.decompress(Response.payload)return content, content_type

​ extract_content函数会接受一段 HTTP 响应数据（Response），以及我想提取的数据类型的名字作为参数。这段响应数据是一个命名元组，里面有两部分：header 和 payload。

​ 如果检测到响应数据被 gzip 或 deflate 之类的工具压缩过，就调用 zlib 库来解压这段数据。任何一个含有图片的响应包，其数据头的 Content-Type 属性里都会有 image 字样（图 3-3）。遇到这种数据头，我就创建一个 content_type 变量，将数据头里指定的实际数据类型保存下来。我还创建另一个变量 content 来保存数据内容，也就是 payload 中 HTTP 头之后的全部数据。最后，将 content 和 content_type 打包成一个元组返回。

图 3-3 Content-Type 字段

​ 写完这两个辅助函数后，就可以开始编写 Recapper 类了：

class Recapper:def __init__(self, fname):pcap = rdpcap(fname)self.sessions = pcap.sessions()self.responses = list()

​ 首先，初始化这个对象，把要读取的 pcap 文件路径传给它。接着我用到 Scapy 的一个美妙功能，自动切分每个 TCP 会话，并保存到一个字典里，字典里面的每个会话都是一段完整的 TCP 数据流。最后，创建一个名为 responses 的空列表，之后我会把在 pcap 文件中读到的响应填进去。

​ get_responses函数会遍历整个 pcap 文件，将找到的每个单独的 Response 都填入刚才的 responses 列表：

def get_responses(self):for session in self.sessions:payload = b''for packet in self.sessions[session]:try:if packet[TCP].dport == 80 or packet[TCP].sport == 80:payload += bytes(packet[TCP].payload)except IndexError:sys.stdout.write('x')sys.stdout.flush()if payload:header = get_header(payload)if header is None:continueself.responses.append(Response(header=header, payload=payload))

​ get_responses函数先遍历整个 sessions 字典中的每个会话，以及每个会话中的每个数据包。然后过滤这些数据，只处理发往 80 端口或者从 80 端口接收的数据。接着，把从所有流量中读取到的数据载荷，拼接成一个单独的名为 payload 的缓冲区。这个操作相当于在 Wireshark 中右键单击一个数据包，选择”Follow TCP Stream“选项（图 3-4）。如果没能成功地拼接 payload 缓冲区（最有可能的情况是，数据包里没有出现 TCP 数据），就在屏幕上打印一个 ”x" 然后继续。

图 3-4 Follow TCP Steram

​ 重组 HTTP 数据后，如果 payload 缓冲区里有数据，就把它交给解析 HTTP 头的函数 get_header，它能帮我逐个检查 HTTP 头的内容。然后，把构造出的 Response 对象附加到 responses 列表里。

​ 最后，遍历整个 responses 列表，如果发现任何含有图片的响应，就用 write 函数将这些图片写到磁盘上：

def write(self, content_name):for i, response in enumerate(self.responses):content, content_type = extract_content(response, content_name)if content and content_type:fname = os.path.join(OUTDIR, f'ex_{i}.{content_type}')print(f'Writeing {fname}')with open(fname, 'wb') as f:f.write(content)

​ 由于我已经提取完所有响应，所以 write 函数只需要遍历这些响应，提取其中的内容，并将内容写到一个文件里就可以了。这个文件会被创建到指定的输出目录里，文件名由 enumerate 函数提供的计数和 content_type 两个值拼接而成，例如 ex_2.jpg 就是一个可能出现的图片文件名。当我运行这个程序时，会创建一个 Recapper 对象，调用它的 get_responses 函数来搜索 pcap 文件中的所有响应，然后将提取出的图片写入磁盘。图 3-5。图 3-6.

图 3-5 程序执行样式

图 3-5 提取的图片

最后

​ 当然，你也可以进一步拓展这个程序，让它不仅能从 pcap 文件中提取图片。

参考

• Scapy
• HTTP 标头