ラベル mosquitto の投稿を表示しています。 すべての投稿を表示
ラベル mosquitto の投稿を表示しています。 すべての投稿を表示

2025年7月28日月曜日

paho-mqtt でタイムアウトを設定する方法

paho-mqtt でタイムアウトを設定する方法

概要

前回 paho-mqtt v2 の使い方を紹介しました
今回は connect と publish 時のタイムアウトを設定する方法を紹介します

環境

  • macOS 15.5
  • Python 3.12.11
  • paho-mqtt 2.1.0

サンプルコード

  • vim app.py
from paho.mqtt.client import Client
from paho.mqtt.enums import CallbackAPIVersion


class CustomClient(Client):
    def __init__(self):
        super().__init__(callback_api_version=CallbackAPIVersion.VERSION2)

    def custom_on_disconnect(self, client, userdata, flags, reason_code, properties):
        print("on_disconnect")
        print(client)
        print(userdata)
        print(flags)
        print(reason_code)
        print(properties)

    def custom_on_publish(self, client, userdata, mid, reason_code, properties):
        print("on_publish")
        print(client)
        print(userdata)
        print(mid)
        print(reason_code)
        print(properties)
        self.disconnect()


if __name__ == "__main__":
    print("start")
    client = CustomClient()
    client.on_disconnect = client.custom_on_disconnect
    client.on_publish = client.custom_on_publish
    # 接続タイムアウトを60秒に設定
    client.connect_timeout = 60
    # ここで60秒待つことを確認する
    client.connect("127.0.0.1", port=1883, keepalive=60, bind_address="")
    info = client.publish(
        topic="default", payload='{"payload": "hoge"}', qos=0, retain=False
    )
    # wait_for_publish は、publish の完了を待つためのメソッド、ACKを待ち来ない場合はタイムアウトする
    info.wait_for_publish(timeout=60)
    print("end")

解説

内部的には Client クラスの self._connect_timeout で接続時間のタイムアウトを管理しておりデフォルトでは 5.0 秒になっています

socket.create_connection(addr, timeout=self._connect_timeout, source_address=source)

もし上書きしたい場合は setter 用の connect_timeout があるのでこれを使います
もし connect 時に指定したブローカーに接続できない場合には timeout 秒待ってからエラーになります

publish 時のタイムアウトは接続時とは少し違い ACK を受け取るまでのタイムアウトを指定します
基本的に publish は非同期なので送ったら送りっぱなしになります
もし送信できたかの確認をしたい場合は ACK 信号を受け取るようにします
その ACK 信号を受け取るためのメソッドが wait_for_publish になっておりその信号を受け取るまでのタイムアウトを指定することができます

最後に

paho-mqtt でタイムアウトを設定する方法を紹介しました
基本的にブローカーに接続できれば publish はできるはずなので connect 側のタイムアウトを設定すれば OK かなと思います

Subscribe などでブローカーと常に接続しておく必要がある場合は timeout + keepalive で接続を確保するようにしておきましょう

参考サイト

2025年7月27日日曜日

paho-mqtt v2 を使う方法

paho-mqtt v2 を使う方法

概要

過去に paho-mqtt の使い方を紹介しました
その際は v1 でしたが v2 となりいろいろと変わっていたので使い方を紹介します

環境

  • macOS 15.5
  • Python 3.12.11
  • paho-mqtt 2.1.0

サンプルコード

前回同様 Client クラスを継承する形でクライアントを定義しています

  • vim app.py
from paho.mqtt.client import Client
from paho.mqtt.enums import CallbackAPIVersion


class CustomClient(Client):
    def __init__(self):
        super().__init__(callback_api_version=CallbackAPIVersion.VERSION2)

    def custom_on_disconnect(self, client, userdata, flags, reason_code, properties):
        print("on_disconnect")
        print(client)
        print(userdata)
        print(flags)
        print(reason_code)
        print(properties)

    def custom_on_publish(self, client, userdata, mid, reason_code, properties):
        print("on_publish")
        print(client)
        print(userdata)
        print(mid)
        print(reason_code)
        print(properties)
        self.disconnect()


if __name__ == "__main__":
    print("start")
    client = CustomClient()
    client.on_disconnect = client.custom_on_disconnect
    client.on_publish = client.custom_on_publish
    client.connect("localhost", port=1883, keepalive=60, bind_address="")
    client.publish(topic="default", payload='{"payload": "hoge"}', qos=0, retain=False)
    print("end")

変わっていた点

各種コールバックメソッドの引数が変わっています
具体的には flags, reason_code, properties が追加になっています
また Client を初期化する際に callback_api_version という引数が指定できることになっておりここに CallbackAPIVersion.VERSION2 を指定することで引数が追加になったコールバックメソッドを使うことができるようになっています

v2 のコードから Client クラスで on_disconnect がコールバック関数のみを受け取るように変更になっています
なので on_disconnect をオーバライドするのではなくコールバック用のメソッド自体を定義してその後そのコールバック用メソッドを on_disconnect に設定することでコールバックメソッドを設定します

最後に

v1 用のコールバックメソッドは非推奨になっており今後は v2 を使うことになるので今のうちにマイグレーションしておきましょう

参考サイト

2024年11月19日火曜日

mosquitto を docker で起動する方法

mosquitto を docker で起動する方法

概要

公式イメージを使ってブローカーを起動する方法を紹介します
認証などはない設定なので検証目的で使いましょう

環境

  • macOS 15.1
  • docker 27.3.1
  • mosquitto 2.0.20

mosquitto.conf

  • vim mosquitto.conf
listener 1883
allow_anonymous true

起動

  • docker run --rm -it -p 1883:1883 -v $(pwd)/mosquitto.conf:/mosquitto/config/mosquitto.conf eclipse-mosquitto

動作確認

sub

  • docker run --rm eclipse-mosquitto mosquitto_sub -t test/topic -p 1883 -h 192.168.1.48

pub

  • docker run --rm eclipse-mosquitto mosquitto_pub -t test/topic -p 1883 -h 192.168.1.48 -m "hoge"

ログ

1731982775: mosquitto version 2.0.20 starting
1731982775: Config loaded from /mosquitto/config/mosquitto.conf.
1731982775: Opening ipv4 listen socket on port 1883.
1731982775: Opening ipv6 listen socket on port 1883.
1731982775: mosquitto version 2.0.20 running
1731982775: New connection from 172.17.0.1:64478 on port 1883.
1731982775: New client connected from 172.17.0.1:64478 as auto-49D03FC4-BBCD-BB77-6AEE-EED2DFA56542 (p2, c1, k60).
1731982805: New connection from 172.17.0.1:55614 on port 1883.
1731982805: New client connected from 172.17.0.1:55614 as auto-D312CD16-EB8D-8E19-7129-5619E4130433 (p2, c1, k60).
1731982805: Client auto-D312CD16-EB8D-8E19-7129-5619E4130433 disconnected.

最後に

実際に使う場合には ACL などの機能を使ってトピックにアクセス制御を設定しましょう

参考サイト

2023年6月30日金曜日

Python paho-mqtt ライブラリを独自クラスとして定義する

Python paho-mqtt ライブラリを独自クラスとして定義する

概要

mqtt.Client をそのまま使わず継承して使います
その場合 on_connect などをインスタンスメソッドとして定義できます

環境

  • macOS 13.4.1
  • Python 3.11.3
  • paho-mqtt 1.6.1
  • mosquitto 2.0.15

mosquittoの起動

  • brew install mosquitto
  • brew services run mosquitto
  • mosquitto_sub -h localhost -p 1883 -t default

publish するサンプルコード

  • vim ./app.py
from paho.mqtt.client import Client


class CustomClient(Client):
    def __init__(self):
        super().__init__()

    def on_disconnect(self, client, userdata, rc):
        print("on_disconnect")
        print(client)
        print(userdata)
        print(rc)

    def on_publish(self, client, userdata, mid):
        print("on_publish")
        print(client)
        print(userdata)
        print(mid)
        self.disconnect()


if __name__ == "__main__":
    print("start")
    client = CustomClient()
    client.connect("localhost", port=1883, keepalive=60, bind_address="")
    client.publish(topic="default", payload='{"payload": "hoge"}', qos=0, retain=False)
    print("end")
  • pipenv run python app.py
  • mosquitto_sub -h localhost -p 1883 -t default

subscribe するサンプルコード

  • vim ./app.py
from paho.mqtt.client import Client


class CustomClient(Client):
    def __init__(self):
        super().__init__()

    def on_connect(self, client, userdata, flags, rc):
        print("on_connect")
        print(client)
        print(userdata)
        print(flags)
        print(rc)

    def on_disconnect(self, client, userdata, rc):
        print("on_disconnect")
        print(client)
        print(userdata)
        print(rc)

    def on_message(self, client, userdata, message):
        print("on_message")
        print(client)
        print(userdata)
        print(message.topic)
        print(message.payload)
        self.disconnect()

    def on_subscribe(self, client, userdata, mid, granted_qos):
        print("on_subscribe")
        print(client)
        print(userdata)
        print(mid)
        print(granted_qos)


if __name__ == "__main__":
    print("start")
    client = CustomClient()
    client.connect("localhost", port=1883, keepalive=60, bind_address="")
    client.subscribe("default", qos=0)
    client.loop_forever()
    print("end")

loop_forever (loop_start/loop_stop) はメッセージを受信するときに使います
publish するときには不要です

  • pipenv run python app.py
  • mosquitto_sub -h localhost -p 1883 -t default
  • mosquitto_pub -h localhost -p 1883 -t default -m "hoge"

最後に

クラス化したほうが管理は楽になるかなと思います

参考サイト

2017年8月7日月曜日

mosquitto-auth-plugin の HTTP 用の認証サーバを作成してみた

概要

前回 mosquitto-auth-plugin をビルドして mosquitto にインストールするところまでやってみました
今回は実際に認証サーバを作成して pub/sub 時の挙動を確認してみたいと思います

環境

  • mosquitto
    • Ubuntu 16.04
    • mosquitto 1.4.8
    • mosquitto-auth-plugin 6eea8229a230aff562906c7bdb8b7a959bd46418
  • 認証サーバ
    • CentOS 7.3.1611
    • Ruby 2.3.3p222
    • sinatra 2.0.0

認証サーバの作成

3 つの URI を POST で受け付けるサーバを作成するだけです
今回は Ruby + Sinatra で作成してみました

  • bundle init
  • vim Gemfile
gem "sinatra"
  • bundle install –path vender/bundle
  • vim app.rb
require 'sinatra'

set :bind, '0.0.0.0'

post '/auth' do
  puts '/auth'
  puts params
  status 200
end

post '/superuser' do
  puts '/superuser'
  puts params
  status 400
end

post '/acl' do
  puts '/acl'
  puts params
  status 200
end
  • bundle exec ruby app.rb

で起動までしましょう
ちょろっと解説すると 3 つの URI は前回 mosquitto の設定ファイルに記載した

...
auth_opt_http_getuser_uri /auth
auth_opt_http_superuser_uri /superuser
auth_opt_http_aclcheck_uri /acl
...

になります
なので設定ファイル側に記載した URI が異なる場合は認証サーバ側で受ける URI の定義も変更してください

そしてレスポンスコードですが 200 or 400 を返却してます
auth-plugin の説明を見ると 200 系が返れば成功みなし、400 系や 500 系が返ると失敗とみなすようです

今回は /auth と /acl は 200 を返却し /superuser は 400 を返すようにしました
こうすることで /auth -> /superuser -> /auth すべての URI がコールされるようになります
もし /superuser で 200 が返ってしまうとその後の /acl のチェックはスルーされるようです
(スーパーユーザだから権限チェックする必要はないということだと思います)

また、今回の実装ではパラメータの情報をデバッグしているだけですが実際には認証のチェックや権限のチェックをしなければなりません
例えば /auth では MySQL に接続して params の情報を元にユーザ認証する機能を持たせるとか、/acl では params の acc 情報を元に該当するユーザがトピックにアクセスできる権限があるかチェックする機能を実装する必要があります

一応公式でも python の参照実装が用意されているのでこれを参考に実装しても良いと思います
https://github.com/jpmens/mosquitto-auth-plug/blob/master/examples/http-auth-be.py

動作確認

auth-plugin を有効にした mosquitto に実際に publish と subscribe をしたみました
その時の認証サーバ側のログも追ってみます
今回の実装ではどんなユーザ/パスワードでも認証を通ります

まずは subscribe してみます

  • mosquitto_sub -d -t topic -u usernameA -P passwordA
/auth
{"username"=>"usernameA", "password"=>"passwordA", "topic"=>"", "acc"=>"-1", "clientid"=>""}
192.168.100.105 - - [04/Aug/2017:11:31:48 +0900] "POST /auth HTTP/1.1" 200 - 0.0004
192.168.100.105 - - [04/Aug/2017:11:31:48 JST] "POST /auth HTTP/1.1" 200 0
- -> /auth

通るのは auth だけのようです
clientid は /auth では送られてこないようです

subscribe できたら publish してみます

  • mosquitto_pub -d -t topic -u usernameA -P passwordA -m “$(date)”
/auth
{"username"=>"usernameA", "password"=>"passwordA", "topic"=>"", "acc"=>"-1", "clientid"=>""}
192.168.100.105 - - [04/Aug/2017:11:32:36 +0900] "POST /auth HTTP/1.1" 200 - 0.0004
192.168.100.105 - - [04/Aug/2017:11:32:36 JST] "POST /auth HTTP/1.1" 200 0
- -> /auth
/superuser
{"username"=>"usernameA", "password"=>"", "topic"=>"", "acc"=>"-1", "clientid"=>""}
192.168.100.105 - - [04/Aug/2017:11:32:36 +0900] "POST /superuser HTTP/1.1" 400 - 0.0003
192.168.100.105 - - [04/Aug/2017:11:32:36 JST] "POST /superuser HTTP/1.1" 400 0
- -> /superuser
/acl
{"username"=>"usernameA", "password"=>"", "topic"=>"topic", "acc"=>"2", "clientid"=>"publisher"}
192.168.100.105 - - [04/Aug/2017:11:32:36 +0900] "POST /acl HTTP/1.1" 200 - 0.0003
192.168.100.105 - - [04/Aug/2017:11:32:36 JST] "POST /acl HTTP/1.1" 200 0
- -> /acl
/superuser
{"username"=>"usernameA", "password"=>"", "topic"=>"", "acc"=>"-1", "clientid"=>""}
192.168.100.105 - - [04/Aug/2017:11:32:36 +0900] "POST /superuser HTTP/1.1" 400 - 0.0003
192.168.100.105 - - [04/Aug/2017:11:32:36 JST] "POST /superuser HTTP/1.1" 400 0
- -> /superuser
/acl
{"username"=>"usernameA", "password"=>"", "topic"=>"topic", "acc"=>"1", "clientid"=>"subscriber"}
192.168.100.105 - - [04/Aug/2017:11:32:36 +0900] "POST /acl HTTP/1.1" 200 - 0.0003
192.168.100.105 - - [04/Aug/2017:11:32:36 JST] "POST /acl HTTP/1.1" 200 0
- -> /acl

5 つほどログが流れました
はじめの /auth, /superuser はおそらく publisher のものだと思います
/auth, /superuser の場合も clientid がないようです
そして、その後の /acl で初めて clientid が付与されます
acc=2 が publish のようです

publisher のあと /superuser と /acl がコールされています
これはおそらく subscriber 側がコールしたものになります
/auth と通っていないのはすでに認証に成功しているためです
subscriber でも clientid が付与されるのは /acl のときだけのようです
subscribe の場合 acc は 1 のようです
/superuser が 400 の場合 /acl のチェックに移行しここで 200 が返ってくれば無事 subscribe することができます

1 つ publish したあとで続けざまに pub してみましょう
すると /auth が 1 度だけコールされます

/auth
{"username"=>"usernameA", "password"=>"passwordA", "topic"=>"", "acc"=>"-1", "clientid"=>""}
192.168.100.105 - - [04/Aug/2017:11:42:50 +0900] "POST /auth HTTP/1.1" 200 - 0.0004
192.168.100.105 - - [04/Aug/2017:11:42:50 JST] "POST /auth HTTP/1.1" 200 0
- -> /auth

これは publisher の /auth になります
/superuser と /acl がチェックされていないのは auth-plugin の機能で /acl チェックをキャッシュする機能があるためです
acl_cacheseconds というパラメータがありデフォルトでは 300 秒に設定されているようです
なので、この時間を過ぎるまでは同一ユーザでのアクセスは /superuser, /acl のチェックがスルーされます
0 に設定することで disable にすることもできます

また、/auth にもキャッシュするがあるようで一度認証が通った場合は認証情報をスルーできます
公式を見るとデフォルトが「0:disable」で設定されているのでコマンドでアクセスする度に毎回 /auth は通ります

おそらくキャッシュ機能は認証サーバ側の負荷を下げるために用意された機能なのでアクセスが多い場合などはキャッシュ有効にすると良いと思います

最後に

auth-plugin の HTTP 機能を使って認証サーバを作成してみました
I/F の部分に関しては実装する部分がそこまで多くないので簡単に作れると思います

ただ、実際に DB などを使って認証、権限チェックをさせる場合いろいろと実装が必要になるのでそこは環境に合わせて変更してください
逆に言うと auth-plugin の HTTP 認証機能の場合、認証するための仕組みを自分で自由に変更できるのは嬉しい点かなと思います

auth-plugin には他にも MySQL や ldap と連携する機能がありますが個人的には HTTP 機能を使ったほうが認証/認可の実装の自由度が高いのでおすすめです
ただ、認証サーバの管理をしなければいけなくなるのでその辺りはトレードオフかなと思います

2017年8月4日金曜日

mosquitto-auth-plugin をビルドして mosquitto にインストールして挙動を確認してみた

概要

mosquitto で認証機構を追加するのに mosquitto-auth-plugin という仕組みがあります
mosquitto は C 言語で書かれているためプラグインも C 言語です
使用するためにはソースからコンパイルする必要があるので今回はその方法を紹介します

環境

  • Ubuntu 16.04
  • make 4.1
  • c-ares 33bf5ba35e2df09ecde9530816733965bb4053e4 (ソースからインストール)
  • openssl 2326bba0e5cbe98f4d00855a6909b1f14b6f5427 (ソースからインストール)
  • mosquitto 1.4.14 (ソースからインストール)
  • mosquitto 1.4.8 (プラグイン確認用)

apt でインストールできるものを事前にインストール

  • apt -y update
  • apt-get -y install make git vim libcurl4-openssl-dev libwrap0-dev build-essential libtool m4 automake uuid-dev

mosquitto のビルドとインストール

mosquitto をビルドしてインストールするには apt でインストールしたモジュール以外に c-ares, openssl が必要になります

c-ares のインストール

これをしないと ares.h が見つからず auth-plugin のビルドでエラーになります

openssl のインストール

これをしないと opensslconf.h が見つからず auth-plugin のビルドでエラーになります

mosquitto のインストール

mosquitto-auth-plug のビルド

# Select your backends from this list
BACKEND_CDB ?= no
BACKEND_MYSQL ?= no
BACKEND_SQLITE ?= no
BACKEND_REDIS ?= no
BACKEND_POSTGRES ?= no
BACKEND_LDAP ?= no
BACKEND_HTTP ?= yes
BACKEND_JWT ?= no
BACKEND_MONGO ?= no
BACKEND_FILES ?= no

# Specify the path to the Mosquitto sources here
# MOSQUITTO_SRC = /usr/local/Cellar/mosquitto/1.4.12
MOSQUITTO_SRC =/root/mosquitto

# Specify the path the OpenSSL here
OPENSSLDIR = /root/openssl

# Specify optional/additional linker/compiler flags here
# On macOS, add
#       CFG_LDFLAGS = -undefined dynamic_lookup
# as described in https://github.com/eclipse/mosquitto/issues/244
#
# CFG_LDFLAGS = -undefined dynamic_lookup  -L/usr/local/Cellar/openssl/1.0.2l/lib
# CFG_CFLAGS = -I/usr/local/Cellar/openssl/1.0.2l/include -I/usr/local/Cellar/mosquitto/1.4.12/include
CFG_LDFLAGS =
CFG_CFLAGS =
  • make clean
  • make

で「auth-plug.so」が作成されていれば OK です

プラグインを使ってみる

今回は別途 Ubuntu に apt で インストールした mosquitto (1.4.8) にプラグインを有効にして使ってみます

  • touch /etc/mosquitto/conf.d/auth.conf
  • vim /etc/mosquitto/conf.d/auth.conf
auth_plugin /root/mosquitto-auth-plug/auth-plug.so
auth_opt_backends http
auth_opt_http_ip 127.0.0.1
auth_opt_http_port 3000
auth_opt_http_getuser_uri /auth
auth_opt_http_superuser_uri /superuser
auth_opt_http_aclcheck_uri /acl
  • systemctl restart mosquitto
  • mosquitto_sub -d -t topic -u username -P password
Client mosqsub|24842-ubuntu sending CONNECT
Client mosqsub|24842-ubuntu received CONNACK
Connection Refused: not authorised.
Client mosqsub|24842-ubuntu sending CONNECT
Client mosqsub|24842-ubuntu received CONNACK
Connection Refused: not authorised.
Client mosqsub|24842-ubuntu sending CONNECT
Client mosqsub|24842-ubuntu received CONNACK
Connection Refused: not authorised.

という感じで認証エラーになればプラグインの有効化は完了です

最後に

mosquitto-auth-plugin をビルドして mosquitto に設定するところまでやってみました
自力でビルドする必要があり、ビルドするのに依存するモジュールもビルドしなければならないのが辛かったです

もしかするとバージョンがあがったりするとビルドの方法が変わるかもしれません
その場合はソースコード内に含まれる README.md や INSTALL.md, compiling.txt などを見れば依存するモジュールやビルド方法が記載されているのでそちらを参照してください

自分もはじめは web でいろいろ調べていたんですが、ビルドの情報が結構古く、結局公式のドキュメントを見るのが一番良かったです

次回は認証サーバを独自で構築して実際に認証できるか試してみます

参考サイト