Impacket mqtt_check.py

概要

MQTT ブローカーに CONNECT を送り、匿名接続の可否と応答内容を確認します。

項目	内容
実装	`examples/mqtt_check.py`
カテゴリ	MSSQL / Exchange / MQTT
規模	84 行
配布	`setup.py` の `scripts=examples/*.py` によりスクリプトとしてインストール

CLI と入力

オプションの最新の定義は python examples/mqtt_check.py -h で確認できます。

引数	説明
`--help`	show this help message and exit
`target`	[[domain/]username[:password]@]
`-client-id`	Client ID used when authenticating (default random)
`-ssl`	turn SSL on
`-port`	port to connect to (default 1883)
`-debug`	Turn DEBUG output ON
`-ts`	Adds timestamp to every logging output

詳細な処理フロー

target、client ID、port、TLS option から socket を作り、MQTT CONNECT packet を送信します。CONNACK を受信・parse し、return code から接続可否を表示して socket を閉じます。

仕組みの詳細

MQTT client は最初の packet として CONNECT を送り、broker は CONNACK で protocol acceptance と authentication result を返します。このコマンドは username/password 認証を試す tool ではなく、client ID と clean-session 等を含む最小 CONNECT で anonymous 接続の可否を確認します。

MQTT fixed header の remaining length は可変長 encoding で、CONNECT payload には protocol name/version、connect flags、keepalive、UTF-8 client ID が順番に入ります。impacket.mqtt の structure がこれを組み立て、socket/TLS 経路が指定 port へ送ります。受信時は少なくとも完全な CONNACK frame が揃うまで読み、return code を解釈します。

TCP 接続成功、TLS handshake 成功、MQTT protocol acceptance、anonymous authorization は別の段階です。旧/new MQTT version 不一致や identifier rejected を「authentication required」と誤解しないようにします。短い packet、過大な remaining length、broker による即時 close も明示的な結果として扱います。

主要な実装要素

シンボル	位置	責務を追う入口
`MQTT_LOGIN`	32	`__init__()`、`run()`

依存関係

Impacket 内部

モジュール	使用する主な名前
`impacket`	`version`
`impacket.examples`	`logger`
`impacket.examples.utils`	`parse_target`
`impacket.mqtt`	`CONNECT_ACK_ERROR_MSGS`, `MQTTConnection`

出力・副作用・必要権限

照会や認証の試行が中心です。実行回数が多いと監査ログやアカウントロック、対象の負荷に影響します。

失敗経路と後始末

入力ファイルがない場合、壊れている場合、読み取りが途中で終わる場合、未対応の形式だった場合を区別します。

実装を変更するとき

impacket.mqtt と標準 socket/TLS に依存し、MQTT_LOGIN.run() が packet exchange を所有します。MQTT remaining-length と短い/不正な応答を検証し、TLS handshake error と broker rejection を区別してください。

検証

構文: python -m py_compile examples/mqtt_check.py
CLI: python examples/mqtt_check.py -h
このコマンド専用のテストは見つかりません。共有モジュールのテストと、必要に応じて隔離環境での動作確認を選びます。