Sparkplug B MQTT Security Fuzzer

Sparkplug B MQTTプロトコル実装をテストするための包括的なセキュリティ評価ツールです。このファザーは、全9つのメッセージタイプにわたるすべてのプロトコルフィールドを体系的にテストし、ネットワーク上の稼働中のデバイスを発見し、詳細なログを分析用に生成します。

責任ある使用

このツールは、ターゲットブローカーに対して、不正な形式、インジェクション、プロトコル違反のMQTTメッセージを送信します。所有しているシステム、または明示的な書面によるテスト許可を得ているシステムに対してのみ実行してください。 Sparkplug Bブローカーは通常、OT/ICS環境に設置されており、予期しないペイロードが物理プロセスを混乱させる可能性があります。明らかにそうでないと証明されない限り、すべてのターゲットは本番環境に隣接していると想定してください。

このツールを使用してSparkplug B実装に脆弱性を発見した場合は、影響を受けるベンダーと協調的な開示を行ってください。このツール自体のセキュリティ問題を報告するには、SECURITY.mdを参照してください。

概要

Sparkplug B仕様は、産業用IoT（IIoT）環境向けにMQTTとGoogle Protocol Buffers上に構築されたトピック名前空間とペイロード形式を定義しています。このファザーは、以下の方法でSparkplug B実装のセキュリティと堅牢性を評価します。

19すべてのメトリックデータタイプを境界値とオーバーフロー条件でテスト
悪意のある文字列の注入（XSS、SQLi、フォーマット文字列、パストラバーサル、コマンドインジェクション）
宣言されたデータタイプと実際のprotobuf値フィールド間の型の不一致の作成
プロトコル状態機械の順序違反（birth前のdata、二重birth、death後のdata）
シリアル化されたprotobufペイロードのバイナリレベルでの破損
検出されたネットワークデバイスのbirth/death証明書のスプーフィング
特殊文字、大文字小文字のバリエーション、構造違反によるMQTTトピック名前空間のファジング

前提条件

Python 3.8以上
MQTTブローカー — テスト対象のシステム（例：Mosquitto、HiveMQ、EMQX、またはSparkplug B対応の任意のブローカー）
許可 — このツールは許可されたセキュリティテスト専用です

インストール

最近のDebian/Ubuntu/Kali（PEP-668システム）では、--setupを使用してシステムPythonにpip installすることはできません。最初に仮想環境または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

手動セットアップ（もし --setup が機能しない場合）

```bash pip install -r requirements.txt git clone https://github.com/eclipse/tahu.git cp tahu/python/core/sparkplug_b.py . cp tahu/python/core/array_packer.py . protoc --python_out=. sparkplug_b.proto rm -rf tahu ```

クイックスタート```bash

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]

Option	Default	Description
`-H`, `--host`	`localhost`	MQTTブローカーのホスト名またはIP
`-p`, `--port`	`1883`（`--tls` 付きの場合は `8883`）	MQTTブローカーのポート
`-u`, `--username`	なし	MQTTユーザー名（`MQTT_USERNAME` 環境変数も読み取ります）
`-P`, `--password`	なし	MQTTパスワード（`MQTT_PASSWORD` も読み取ります。`-` を指定するとエコーなしで標準入力から読み取ります）
`--tls`	off	TLSで接続します。`-p` が未設定の場合、デフォルトポートは `8883` になります
`--cafile`	なし	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のパブリッシュを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`	—	追加の文字列インジェクションペイロードファイルへのパス（1行1ペイロード、UTF-8）。組み込みの `STRING_FUZZ_VALUES` に追加されます（置き換えません）。最大10 MB / 10,000ペイロード。カスタム文字列コーパスを参照

ファズカテゴリ

Category	Description	Approx. Cases
`boundary`	全19の数値データ型の最小/最大/オーバーフロー、値付きis_null、フラグの組み合わせ	~200
`string`	String、Text、UUID、MetaDataフィールド、STATEメッセージ全体でのインジェクションペイロード（XSS、SQLi、フォーマット文字列、パストラバーサル、コマンドインジェクション、nullバイト）	~100
`type_mismatch`	宣言されたデータ型と間違ったprotobuf値フィールド、無効なデータ型コード、複数のoneofフィールド	~150
`sequence`	シーケンスの欠落、重複、逆順、巻き戻し、NBIRTH/NDEATH間のbdSeqの不一致	~20
`timestamp`	ゼロ、最大uint64、遠い未来/過去、メトリックとペイロードタイムスタンプの不一致、DateTimeの極端な値	~15
`alias`	異なるメトリックに対するエイリアスの重複、極端なエイリアス値、データメッセージ内の未定義エイリアス	~15
`orphan`	存在しないデバイス、ノード、グループをターゲットにしたデータ/コマンド、未定義のテンプレート参照	~20
`ordering`	プロトコル状態違反：Birth前のデータ、二重Birth、Death後のデータ、間違ったBirth順序	~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

System trust store, default port 8883

python3 sparkplug-fuzzer.py -H broker.example.com --tls -v

Custom CA bundle

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

カスタムグループ/ノードのIDとログファイル:```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

transfer ~/tahu to the air-gapped target, then on the target:

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

sparkplugFuzzer

人気