MQTT Broker mosquito配置以及使用tls证书登录

1. 前言

Eclipse Mosquitto是一个开源消息代理，实现了MQTT协议版本3.1和3.1.1。Mosquitto轻量，适用于低功耗单板计算机到完整服务器的所有设备。Mosquitto项目还提供了用于实现MQTT客户端的C库以及非常受欢迎的mosquitto_pub和mosquitto_sub命令行MQTT客户端。

2. 安装

ubuntu下面使用apt就可以快速安装mosquito。

sudo apt-get update
sudo apt-get install mosquitto

安装客户端

sudo apt-get install mosquitto-clients

安装完成以后，mosquito的配置文件会在/etc/mosquitto目录下面。

root@ubuntu:/etc# cd mosquitto/
root@ubuntu:/etc/mosquitto# ll
total 52
drwxr-xr-x   5 root root  4096 Nov 10 11:27 ./
drwxr-xr-x 137 root root 12288 Nov 10 08:44 ../
-rw-r--r--   1 root root    89 Nov  2 17:41 aclfile
-rw-r--r--   1 root root   230 Feb 28  2020 aclfile.example
drwxr-xr-x   2 root root  4096 Nov 10 09:47 ca_certificates/
drwxr-xr-x   2 root root  4096 Nov 10 09:47 certs/
drwxr-xr-x   2 root root  4096 Nov  2 14:36 conf.d/
-rw-r--r--   1 root root   597 Nov 10 11:27 mosquitto.conf
-rw-r--r--   1 root root    23 Feb 28  2020 pskfile.example
-rw-r--r--   1 root root   226 Nov  2 15:43 pwfile
-rw-r--r--   1 root root   355 Feb 28  2020 pwfile.example

mosquito的log文件则在/var/log/mosquitto目录

root@ubuntu:/etc/mosquitto# cd /var/log/mosquitto/
root@ubuntu:/var/log/mosquitto# ll
total 12
drwxr-xr-x  2 mosquitto root      4096 Nov 10 15:37 ./
drwxrwxr-x 15 root      syslog    4096 Nov 10 13:56 ../
-rw-------  1 mosquitto mosquitto 1831 Nov 10 16:12 mosquitto.log
root@ubuntu:/var/log/mosquitto# 

3. mosquito相关命令

3.1 运行停止查看状态

sudo service mosquitto status
sudo service mosquitto start
sudo service mosquitto stop

3.2 创建可以登录mosquito的用户

我们需要指定登录mosquito的用户，这里需要通过命令行来创建用户，运行前，需要检查pwfile是否存在，如果不存在，就创建。

touch /etc/mosquitto/pwfile

mosquitto_passwd /etc/mosquitto/pwfile 用户名

其中用户名处，需要输入一个用户名，然后我们根据命令提示，根据提示输入两遍密码。创建用户完成以后，我们就可以看到pwfile里面保存的用户信息如下。

root@ubuntu:/etc/mosquitto# cat pwfile
pub:$6$tsyy9ESjhVYZbz1R$91a8CIsE164xWgjXFvE66w4UJ3LzId4CE8W/9EwJqFXUFAm4eRAytqjJmphI9upMxi1VRcJbDoNl61MuCkThxw==
sub:$6$UPg6lTVxYodbsKwa$CDUTJOmZUrDykCbNZO0IlRE15a1uJe6G8QxehLeACiVAx2kWwY43Lz5EXF9m8XtBEFltfbfQMHQkpGnWjilBGg==

这里定义了pub和sub两个用户，分别表示发送者以及接收者，当然后面的密码也是经过加密的。

3.3 配置权限规则文件

我们刚刚创建了2个登录用户，接下来，我们需要定义这两个用户的订阅规则，包括收发的权限。这里我们修改/etc/mosquitto/aclfile，如果没有这个文件，就创建。

touch /etc/mosquitto/aclfile
vim /etc/mosquitto/aclfile

文件内容如下：

user pub
topic write testuser sub
topic read test

上面的的内容规定了，用户pub只能够发送topic为test的消息。用户sub只能订阅topic为test的消息。即一个是负责发送消息，一个是负责接收消息。

3.4 配置mosquito

最后，我们需要把刚刚修改的配置文件，加入到mosquito的conf文件中，分别添加了用户文件，规则文件，以及进制了匿名用户的登录。

vim /etc/mosquitto/mosquitto.conf

配置文件如下，其中adding后面，是这里需要添加的内容。

# Place your local configuration in /etc/mosquitto/conf.d/
# 
# A full description of the configuration file is at
# /usr/share/doc/mosquitto/examples/mosquitto.conf.examplepid_file /var/run/mosquitto.pidpersistence true
persistence_location /var/lib/mosquitto/log_dest file /var/log/mosquitto/mosquitto.loginclude_dir /etc/mosquitto/conf.d# adding
allow_anonymous false
password_file /etc/mosquitto/pwfile
acl_file /etc/mosquitto/aclfile

最后我们重启mosquito，就可以完成了配置。

service mosquito restart

3.5 使用mosquito_pub和mosquito_sub测试

我们可以再shell中使用mosquito_pub和mosquito_sub来测试刚刚的配置是否成功。

3.5.1 监听消息

这里监听了topic为test的信息。用户名和password都是刚刚定义的。

mosquitto_sub -h 192.168.0.120 -t "test" -u sub -P 123456 -i "client1" 

3.5.2 发送消息

mosquitto_sub -h 192.168.0.120 -t "test" -u sub -P 123456 -i "client1"

4. 使用TLS证书登录

如果单纯使用用户名和密码登录，可能会有安全相关问题。为了解决此问题，我们可以用TLS签发证书配合mosquito进行安全校验，证书的签发以及mosquito的配置主要分成以下这10步。以下是我从参考网站上面截图下来的，每一步可以直接把命令复制进去即可。

TLS证书签发以及mosquito登录配置

Overview of Steps

1. Create a CA key pair
2. Create CA certificate and use the CA key from step 1 to sign it.
3. Create a broker key pair don’t password protect.
4. Create a broker certificate request using key from step 3
5. Use the CA certificate to sign the broker certificate request from step 4.
6. Now we should have a CA key file,a CA certificate file, a broker key file, and a broker certificate file.
7. Place all files in a directory on the broker e.g. certs
8. Copy the CA certificate file to the client.
9. Edit the Mosquitto conf file to use the files -details below
10. Edit the client script to use TLS and the CA certificate. -details below

4.1 创建一个key

key是TLS机制中最重要的东西。是用来生成后面的证书的，创建key时候可以使用密码来保护你的key，key创建后需要妥善保存。

我们创建一个长度为2048的key。

openssl genrsa -des3 -out ca.key 2048

这里我们需要用到openssl的库，可以直接用apt来安装。创建时候需要输入2次密码，输入后要保存好密码，这里我们生成了一个ca.key

4.2 创建一个客户端证书

这里需要创建一个证书给客户端使用，并用刚刚生成的key来签发。注意我们一个key可以生成很多个证书。

openssl req -new -x509 -days 1826 -key ca.key -out ca.crt

上面的命令执行的结果如下，我们签发了一个ca.crt的证书，注意按照命令提示，首先需要输入key的保护密码，然后写一些地址等信息，注意common name这一个字段，需要写上本机的IP地址。

4.3 生成mosquito服务器使用的key

执行下面的命令，生成服务器需要使用的key。这一步就不需要指定密码，这里生成了server.key

openssl genrsa -out server.key 2048

4.4 创建一个用于请求的csr文件

我们用刚刚创建的服务器的key来生成一个用于请求的csr文件。

openssl req -new -out server.csr -key server.key

这里同样需要填入一些基础信息，重要的还是common name这一个字段，需要写上本机的IP地址。并且还有一个challenge password，我也不太理解他是干嘛的，我设成和客户端key一样，方便记忆。
执行完成以后，就生成了server.csr文件。

4.5 签发server的证书

这一步是最重要的了，我们需要用ca.crt, ca.key, server.crt一用来签发server的证书，命令如下。

openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 360

这里还是需要输入ca.key的保护密码来签发，完成后，就可以生成ca.srl以及server。crt文件了。

4.6 验证文件

在进入mosquito配置前，我们需要先验证刚刚生成一系列文件是否有效。

openssl verify -CAfile ca.crt server.crt

验证成功后说明前面文件是确实有效的。

4.7 复制文件到指定的目录

其实我们刚刚生成的文件需要用到的就只有3个文件而已。分别是ca.crt, server.key, server.crt

我们把server.crt, server.key复制到/etc/mosquitto/certs目录，把ca.crt复制到/etc/mosquitto/ca_certificates目录，就可以完成文件部署。

4.8 配置mosquito.conf

再次打开mosquito.conf进行配置，加上cafile, keyfile, certfile, tls_version配置。

vim /etc/mosquitto/mosquitto.conf

# Place your local configuration in /etc/mosquitto/conf.d/
# 
# A full description of the configuration file is at
# /usr/share/doc/mosquitto/examples/mosquitto.conf.examplepid_file /var/run/mosquitto.pidpersistence true
persistence_location /var/lib/mosquitto/log_dest file /var/log/mosquitto/mosquitto.loginclude_dir /etc/mosquitto/conf.d# adding
allow_anonymous false
password_file /etc/mosquitto/pwfile
acl_file /etc/mosquitto/aclfilecafile /etc/mosquitto/ca_certificates/ca.crt
keyfile /etc/mosquitto/certs/server.key
certfile /etc/mosquitto/certs/server.crt
tls_version tlsv1.2

重启mosquito

service mosquito restart

5. 使用shell命令进行带证书测试

登录时候，我们需要带上客户端的证书ca.crt。

5.1 监听

mosquitto_sub -h 192.168.0.120 -p 1883 -t "test" -u sub -P 123456 -i "sub_client" --cafile ca.crt --tls-version tlsv1.2

5.2 发送消息

mosquitto_pub -h 192.168.0.120 -p 1883 -t test -m "Hello world" -u pub -P 123456 -i "pub_client" --cafile ca.crt --tls-version tlsv1.2

这里发送了一句hello world，topic依旧是test，发送完成后，如果接收端能够收到Message，就成功了，如下图。

6. Python实现发布端和监听端

Python可以使用paho包实现mqtt消息的监听和发送。这里也附上实现的相关代码。这里需要用到刚刚生成的ca.crt，以及ca.key.
文件目录：

6.1 publisher.py

# -*- coding: utf-8 -*-
import time
import os
import traceback
import ssl
import paho.mqtt.client as mqtt
from config.load_config import load_cfg
from tools.config_logging import create_loggerconfig = load_cfg("config/config.yaml")
logger_path = os.path.join(config["log_dir"], config["publisher"]["log_file_name"])
pub_logger = create_logger(logger_name=config["publisher"]["log_name"], logger_path=logger_path)
conn_flag = Falseclass Publisher(object):def __init__(self, cfg):self.cfg = cfg# publisher配置部分self.publisher_cfg = self.cfg["publisher"]# mqtt broker配置部分self.mqtt_broker_cfg = self.cfg["mqtt_broker"]self.client = mqtt.Client()# 绑定回调函数# 断开连接时候的回调函数self.client.on_disconnect = self.disconnect_callbackself.client.on_publish = self.on_publish_callbackself.client.on_connect = self.connect_callbackself.conn_flag = Falsedef create_context(self):"""客户端登录信息的设置:return:"""context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)context.load_cert_chain(certfile=self.publisher_cfg["certfile"],keyfile=self.publisher_cfg["keyfile"],password=self.publisher_cfg["cert_pass"],)return context@staticmethoddef connect_callback(client, userdata, flags, rc):assert rc == 0global conn_flagconn_flag = Truepub_logger.info("\nMQTT broker connected successfully,\n""address: {},\n""   port: {}\n".format(client._host, client._port))@staticmethoddef disconnect_callback(client, userdata, rc):# rc == 0, 代表断开连接成功assert rc == 0pub_logger.info("\nDisconnect from MQTT broker,\n""address: {},\n""   port: {}\n".format(client._host, client._port))@staticmethoddef on_publish_callback(client, userdata, mid):topic, msg = userdatapub_logger.info("\nSuccessfully published message,\n""[topic]: {}\n""[content]: {}\n".format(topic, msg))def connect_mqtt_broker(self):"""链接mqtt_broker服务器:return:"""# 验证self.client.username_pw_set(self.publisher_cfg["username"], self.publisher_cfg["password"])# 使用context写法context = self.create_context()self.client.tls_set_context(context)# 不用context的写法# self.client.tls_set(ca_certs="cert/ca.crt", tls_version=ssl.PROTOCOL_TLSv1_2)self.client.tls_insecure_set(True)# 连接 IP port keepaliveself.client.connect(self.mqtt_broker_cfg["address"],self.mqtt_broker_cfg["port"],self.mqtt_broker_cfg["keepalive"])while not conn_flag:time.sleep(0.1)self.client.loop()def disconnect_mqtt_broker(self):self.client.disconnect()def publish_msg(self, topic, msg):"""发送一条消息:param topic::param msg::return:"""pub_logger.info("\nPublishing message,\n""[topic]: {}\n""[content]: {}\n".format(topic, msg))self.client.user_data_set((topic, msg))# 发布 topic 内容self.client.publish(topic, payload=msg.encode(), qos=0)def demo():message = "congratulation, this is a mqtt message."try:publisher = Publisher(cfg=config)publisher.connect_mqtt_broker()publisher.publish_msg(topic=config["publisher"]["topic"], msg=message)publisher.disconnect_mqtt_broker()except Exception as e:# 服务启动失败pub_logger.error(traceback.format_exc())raise eif __name__ == '__main__':demo()

6.2 subscriber.py

# -*- coding: utf-8 -*-
import os
import time
import paho.mqtt.client as mqtt
from tools.config_logging import create_logger
from config.load_config import load_cfg
import traceback
import sslconfig = load_cfg("config/config.yaml")
logger_path = os.path.join(config["log_dir"], config["subscriber"]["log_file_name"])
sub_logger = create_logger(logger_name=config["subscriber"]["log_name"], logger_path=logger_path)
conn_flag = Falseclass Subscriber(object):def __init__(self, cfg):"""MQTT消息订阅客户端:param cfg:"""self.cfg = cfg# subscriber配置部分self.subscriber_cfg = self.cfg["subscriber"]# mqtt broker配置部分self.mqtt_broker_cfg = self.cfg["mqtt_broker"]self.client = mqtt.Client()# 绑定回调函数# 连接回调函数self.client.on_connect = self.connect_callback# 断开连接时候的回调函数self.client.on_disconnect = self.disconnect_callback# 收到消息的回调self.client.on_message = self.on_message_callback# 断开连接时候回调self.client.on_disconnect = self.disconnect_callbackself.client.on_subscribe = self.subscribe_callbackdef create_context(self):context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)context.load_cert_chain(certfile=self.subscriber_cfg["certfile"],keyfile=self.subscriber_cfg["keyfile"],password=self.subscriber_cfg["cert_pass"],)return context@staticmethoddef connect_callback(client, userdata, flags, rc):# rc == 0 为成功，否则就失败assert rc == 0global conn_flagconn_flag = Truesub_logger.info("\nMQTT broker connected successfully,\n""address: {},\n""   port: {}\n".format(client._host, client._port))@staticmethoddef disconnect_callback(client, userdata, rc):# rc == 0 为成功，否则就失败assert rc == 0sub_logger.info("\nDisconnect from MQTT broker,\n""address: {},\n""   port: {}\n".format(client._host, client._port))@staticmethoddef subscribe_callback(client, userdata, mid, granted_qos):sub_logger.info("Subscribed successfully.\n")@staticmethoddef on_message_callback(client, userdata, message):msg_bytes = message.payloadmsg_str = msg_bytes.decode()sub_logger.info("\nReceived message:\n""[topic]: {}\n""[content]: {}\n".format(message.topic, msg_str))def connect_mqtt_broker(self):"""链接mqtt_broker服务器:return:"""# 设置用户名和密码self.client.username_pw_set(self.subscriber_cfg["username"], self.subscriber_cfg["password"])# 使用context写法context = self.create_context()self.client.tls_set_context(context=context)# 不用context的写法# self.client.tls_set(ca_certs="cert/ca.crt", tls_version=ssl.PROTOCOL_TLSv1_2)self.client.tls_insecure_set(True)# 连接 IP port keepaliveself.client.connect(self.mqtt_broker_cfg["address"],self.mqtt_broker_cfg["port"],self.mqtt_broker_cfg["keepalive"])while not conn_flag:time.sleep(0.1)self.client.loop()def run_sub(self):"""开始监听mqtt消息:return:"""# 连接brokerself.connect_mqtt_broker()# 订阅的 topicself.client.subscribe(self.subscriber_cfg["topic"], qos=0)sub_logger.info("\nStarting monitoring on broker for following topic: \n""[topic]: {}\n".format(self.subscriber_cfg["topic"]))self.client.loop_forever()if __name__ == '__main__':sub = Subscriber(cfg=config)try:sub.run_sub()except Exception as e:# 服务启动失败sub_logger.error(traceback.format_exc())raise e

6.3 config.yaml

# 通用变量
log_dir: log# mosquitto服务地址
mqtt_broker:address: 192.168.0.120port: 1883keepalive: 30
# 订阅者
subscriber:# 证书相关参数certfile: "cert/ca.crt"keyfile: "cert/ca.key"cert_pass: "123456"# 客户端相关参数username: "sub"password: "123456"topic: "test"log_name: mqtt subscriberlog_file_name: sub.log
# 发布者
publisher:# 证书相关参数certfile: "cert/ca.crt"keyfile: "cert/ca.key"cert_pass: "123456"# 客户端相关参数username: "pub"password: "123456"topic: "test"log_name: mqtt publisherlog_file_name: pub.log

7. 总结

以上就是mosquito的安装配置已经，Python收发mqtt消息的全部内容了，整理不易，如果同学们觉得这个教程还有用，记得点赞加关注哦。