ドキュメント

# はじめの一歩

インストール後、ブローカーは 3 つのポートで待ち受けます。TCP 上の MQTT は :1883、TLS 上の MQTT は :8883、Web ダッシュボードと /mqtt での WebSocket 上の MQTT は :8567 です。

http://localhost:8567 を開き、管理者としてログインしてください。初期のログインは admin / admin です。最初のログインの直後に、プロフィール画面でパスワードを変更してください。

最初のユーザーを作りましょう。名前、8 文字以上のパスワード、そして ACL ルールを 1 つ。$u/# のルールで機器に自分の枝への完全なアクセスを与えられます。ダッシュボードにはそのための「自分の枝」ボタンがあります。

mosquitto_pub -h localhost -p 1883 -u sensor-42 -P s3nsorPass! \
  -t 'sensor-42/temp' -m '21.5'

mosquitto_sub -h localhost -p 1883 -u sensor-42 -P s3nsorPass! \
  -t 'sensor-42/#' -v

# Linux へのインストール

おすすめは APT リポジトリです。更新はシステムの更新と一緒に届きます。

sudo wget -qO /usr/share/keyrings/elx-repo.gpg https://repo.um-d.ru/elx-repo.gpg
echo "deb [signed-by=/usr/share/keyrings/elx-repo.gpg] https://repo.um-d.ru stable main" | sudo tee /etc/apt/sources.list.d/elx-repo.list
sudo apt-get update
sudo apt-get install elxmqttbroker

パッケージはバイナリを /opt/elxmqttbroker に置き、ログインできないシステムユーザー elxmqtt を作り、その隣にデータと config.json（モード 0600）を配置し、elxmqttbroker サービスを起動時に有効化します。

systemctl status elxmqttbroker      # 状態
sudo systemctl restart elxmqttbroker # 再起動
journalctl -u elxmqttbroker -f      # ログ

ユニットは ProtectSystem=strict で締めてあり、書き込めるのはブローカー自身のディレクトリだけです。既定ではダッシュボードは 127.0.0.1:8567 で待ち受けて nginx 経由で公開し、MQTT の 1883 と 8883 は直接公開します。別のプロトコルであり、代理を挟む必要がないからです。

# Windows へのインストール

通常はインストーラー elxmqttbroker-setup.exe を使います。プログラムを Program Files に、データを C:\ProgramData\ELX MQTT Broker に置き、必要に応じて ELXMQTTBroker サービスをシステム起動時に登録し、ファイアウォールでポート 1883、8883、8567 を開けます。表示はロシア語と英語で、管理者権限が必要です。

ポータブル版は単体の elxmqttbroker.exe です。専用のフォルダーに置いて実行すると、隣に config.json、data ディレクトリ、自己署名の TLS 証明書ができます。

elxmqttbroker.exe                    # 既定の設定で起動
elxmqttbroker.exe -web :9000         # ダッシュボードを別のポートで
elxmqttbroker.exe -version           # バージョンとビルド情報

ポータブル版も、インストーラーが使うのと同じコマンドで、手動でサービスとして扱えます：

elxmqttbroker.exe -service install -config "C:\ProgramData\ELX MQTT Broker\config.json" -data "C:\ProgramData\ELX MQTT Broker\data"
elxmqttbroker.exe -service start
elxmqttbroker.exe -service stop
elxmqttbroker.exe -service uninstall

# 設定

config.json はブローカーが読む唯一のファイルです。初回起動時に作られ、起動前に分かっていなければならないものだけを持ちます。アドレス、TLS、管理者のアカウントです。

{
  "dataDir": "data",
  "auth": { "username": "admin", "password": "admin" },
  "mqtt": { "tcp": ":1883", "tls": ":8883" },
  "web": { "addr": ":8567", "tlsAddr": "" },
  "tls": { "cert": "", "key": "", "selfSigned": true,
            "hosts": ["localhost", "127.0.0.1", "::1"] },
  "logLevel": "info",
  "logFormat": "text",
  "persistence": true
}

平文の password は初回起動時に passwordHash へ変換され、ファイルからは消されます。それ以外——プロトコルの上限、SMTP、ブリッジ、ルール、API トークン、MQTT ユーザー——はすべてダッシュボードで設定し、data/users.json に保存されます。いずれも再起動は不要です。

コマンドラインのオプションはファイルの設定より優先されます：

# ユーザーと権限

系統が 2 つあり、混同してはいけません。管理者は config.json で定義される単一のアカウントで、扱うのはダッシュボードだけです。役割の区分も登録機能もなく、入れば全権を持ちます。MQTT デバイスは別のアカウント一覧で、1〜64 文字の名前、パスワード、有効・無効の印、トピック権限を持ちます。

ACL ルールは、トピックフィルター、アクセス種別、判断の 3 つで構成されます：

ルールは上から順に確認され、最初に一致したものが勝ち、既定はすべて拒否です。明示的な許可がなければアクセスできません。順序も設定の一部なので、ルールはマウスでドラッグしたり矢印キーで動かしたりできます。

フィルターの中の $u は、接続してきたクライアント名に展開されます。$u/# というルール 1 本で各機器を自分の枝に閉じ込められ、名前ごとにルールを書く手間が省けます。

パスワードは PBKDF2-HMAC-SHA256、21 万回の反復とランダムなソルトで保存します。セッションと API のトークンもハッシュのみを保持します。

# MQTT の認証

認証情報の取得元は設定 → 認証で選び、有効にできるのは常に 1 つだけです。切り替えは稼働中に行います。新しい取得元をまず「テスト」ボタンで確認し、応答があった場合にのみブローカーが切り替え、各クライアントが認証し直すよう接続を切ります。取得元に届かない場合は何も変わりません。

MySQL と MariaDB のスキーマは EMQX に倣っています：

CREATE TABLE mqtt_user (username VARCHAR(128) PRIMARY KEY, password VARCHAR(255));
CREATE TABLE mqtt_acl  (username VARCHAR(128), action VARCHAR(16),
                        permission VARCHAR(16), topic VARCHAR(255));
-- action: publish | subscribe | pubsub;  permission: allow | deny
INSERT INTO mqtt_user VALUES ('dev', SHA2('devpass',256));
INSERT INTO mqtt_acl  VALUES ('dev','pubsub','allow','dev/#');

"mqttAuth": {
  "backend": "mariadb",
  "sql": {
    "dsn": "mqtt:mqttpass@tcp(127.0.0.1:3306)/mqtt",
    "passwordQuery": "SELECT password FROM mqtt_user WHERE username = ${username} LIMIT 1",
    "aclQuery": "SELECT action, permission, topic FROM mqtt_acl WHERE username = ${username}",
    "passwordHash": "sha256"
  }
}

${username} と ${clientid} は文字列連結ではなくクエリのパラメーターとして束縛されるので、注入は起こりません。passwordHash は plain、sha256、sha512、bcrypt、pbkdf2 を受け付けます。

JWT はパスワードかユーザー名の欄に入れます。HS256/384/512 と RS256/384/512 に対応し、署名と exp、nbf を検証し、必要に応じて iss、aud、独自のクレームも確認します。権限は EMQX のどちらの形式でも acl クレームから読み取り、is_superuser は検査そのものを免除します。

HTTP の取得元は {username, password, clientid, peerhost} を受け取り、{"result": "allow", "acl": […]} で応答します。空の 200 は許可とみなされ、deny、401、403、時間切れ、あらゆるエラーは拒否です。到達できない認可サービスが開いた扉になってはいけません。応答は cacheSeconds の間キャッシュされます。

# ルール：書き換え、自動購読、制限

トピックの書き換えは、トピック名をその場で置き換えます。ルールは、安価な絞り込みのための MQTT フィルター、正規表現、$1…$9 のグループを使える宛先テンプレートで構成します：

順序は EMQX に倣い、まず書き換え、次に ACL の確認です。権限は、メッセージが実際に流れるトピックに対して検査されます。UNSUBSCRIBE も SUBSCRIBE と同じ書き換えを通ります。そうでなければ購読を解除できなくなるからです。

自動購読は、クライアントからの SUBSCRIBE なしに、接続直後に購読を作ります。%c/# や %u/# のルールで、手作業なしに各クライアントへ自分の枝を与えられます。

トピック制限は、クライアント全体としては問題ないのに、特定のトピックだけが多すぎる場合のためのものです：

フラッド制御は設定 → ブローカーで指定し、接続ごとに効きます。毎秒のメッセージ数、メッセージのバースト、毎秒のバイト数、バイトのバースト。0 は無制限を意味します。超過しても接続は切れません。ブローカーは次のソケット読み取りの前に待ち、TCP がウィンドウを閉じ、送信側が自然に速度を落とします。

# 遅延配信

$delayed/{秒}/{トピック} への発行はブローカーが預かり、期限が来たら購読者へ渡します。形式は EMQX 互換です。

mosquitto_pub -t '$delayed/60/sensors/alarm' -m '1 分後に'

間隔は 1 秒から 4,294,967（およそ 49 日）まで。確認応答はすぐに返ります。ブローカーが保証するのは受け取ったことであり、購読者の存在ではありません。分解能は 1 秒です。

待ち行列はルール → 遅延で確認できます。トピック、ペイロード、残り時間が並び、1 件ずつでも一括でも取り消せます。メッセージは再起動を越えて残り、ブローカーが止まっている間に期限が来たものは起動直後に配信されます。

# ブローカー間ブリッジ

ブローカーはクライアントとして別の MQTT ブローカーに接続し、トピックを双方向に転送できます。設定は設定 → ブリッジで行います。接続先のアドレス（ポートは省略可。既定は 1883、TLS では 8883）、認証情報、TLS、プロトコル版、ルールを指定します。

ループ防止、自動再接続、バージョン 5 での No Local（反響も重複もなし）、ダッシュボードでの状態とカウンターを備えます。ブリッジは別のクライアントとしてではなく、ブローカーの水準で動きます。

# REST API

ダッシュボードはセッションの Cookie で自分の API を呼びます。外部のシステムは設定 → API で作るベアラートークンを使います。トークンの表示は 1 度きりで、保存されるのはハッシュだけです。読み取り専用として発行することもでき、その場合は GET、HEAD、OPTIONS だけが通ります。

curl -H "Authorization: Bearer TOKEN" http://localhost:8567/api/stats

典型的な使い方は、自社のシステムが機器を登録し、同時にブローカーへのアクセスを与えることです：

# 作成
curl -X POST http://localhost:8567/api/users \
  -H "Authorization: Bearer TOKEN" -H "Content-Type: application/json" \
  -d '{"username":"sensor-42","password":"s3nsorPass!","enabled":true,
       "acl":[{"filter":"$u/#","access":"readwrite","allow":true}]}'

# パスワードや権限の変更（送った項目だけ）
curl -X PATCH http://localhost:8567/api/users/sensor-42 \
  -H "Authorization: Bearer TOKEN" -H "Content-Type: application/json" \
  -d '{"password":"newPass12345","enabled":false}'

# 削除
curl -X DELETE http://localhost:8567/api/users/sensor-42 \
  -H "Authorization: Bearer TOKEN"

パスワードが返されることはありません。エラーは {"error":{"code":"…","message":"…"}} という単一の形で返ります：duplicate、reserved、invalid_input、not_found、read_only、unauthorized。

変更は即座に効きます。パスワードや権限の変更、削除の際には、ブローカーがそのユーザーの現在の接続を自ら切るので、新しいルールは次の再接続を待たずにすぐ適用されます。

ダッシュボードの API はすべて利用できます。統計と履歴、クライアント、購読、retain メッセージ、発行、イベント、ユーザー、設定、ブリッジ、ルール、遅延メッセージ、インポートとエクスポート。

# 保存と再起動

persistence: true のとき、状態は data/broker.db に置かれます。WAL モードの SQLite で、純 Go、CGO は使いません。再起動を越えて残るものは：

• retain メッセージ — 一定間隔ではなく、変更のたびに即座に書き込みます。Message Expiry を過ぎたものは起動時に捨てられます。
• 永続セッション — 購読、QoS 1/2 のオフラインキュー、未完了のやり取り。10 秒ごとと終了時に保存します。クリーンセッションは保存しません。仕様上、接続とともに消えるものだからです。
• 遅延メッセージ — ブローカーが止まっている間に期限が来たものは、起動直後に配信されます。

起動時には state restored retained=128 sessions=7 subscriptions=34 delayed=2 のような行がログに出ます。その隣、data/ には users.json（アカウント、ルール、ダッシュボードの設定。書き込みは不可分に行われます）と自己署名の証明書があります。

# 既知の制限

• クラスタリングには対応していません。このブローカーは単一プロセスとして設計されています。
• ブリッジが扱うのは QoS 0 と 1 で、QoS 2 は 1 に落とされます。
• SCRAM のような拡張認証はプロトコルにはありますが、内蔵の取得元は提供せず、0x8C Bad authentication method を返します。
• 頻度制限はユーザー単位ではなく接続単位です。10 本の接続を張るクライアントには 10 個分の上限が与えられます。
• マシンの状態は Linux と Windows で算出します。macOS と BSD では、その欄は正直に「データなし」と表示します。