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

文章目录

MQTT Broker mosquito配置以及使用tls证书登录
- 1. 前言
- 2. 安装
- 3. mosquito相关命令
- - 3.1 运行停止查看状态
  - 3.2 创建可以登录mosquito的用户
  - 3.3 配置权限规则文件
  - 3.4 配置mosquito
  - 3.5 使用mosquito_pub和mosquito_sub测试
  - - 3.5.1 监听消息
    - 3.5.2 发送消息
- 4. 使用TLS证书登录
- - 4.1 创建一个key
  - 4.2 创建一个客户端证书
  - 4.3 生成mosquito服务器使用的key
  - 4.4 创建一个用于请求的csr文件
  - 4.5 签发server的证书
  - 4.6 验证文件
  - 4.7 复制文件到指定的目录
  - 4.8 配置mosquito.conf
- 5. 使用shell命令进行带证书测试
- - 5.1 监听
  - 5.2 发送消息
- 6. Python实现发布端和监听端
- - 6.1 publisher.py
  - 6.2 subscriber.py
  - 6.3 config.yaml
- 7. 总结

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 test

user 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.example

pid_file /var/run/mosquitto.pid

persistence true
persistence_location /var/lib/mosquitto/

log_dest file /var/log/mosquitto/mosquitto.log

include_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

Create a CA key pair
Create CA certificate and use the CA key from step 1 to sign it.
Create a broker key pair don’t password protect.
Create a broker certificate request using key from step 3
Use the CA certificate to sign the broker certificate request from step 4.
Now we should have a CA key file,a CA certificate file, a broker key file, and a broker certificate file.
Place all files in a directory on the broker e.g. certs
Copy the CA certificate file to the client.
Edit the Mosquitto conf file to use the files -details below
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.example

pid_file /var/run/mosquitto.pid

persistence true
persistence_location /var/lib/mosquitto/

log_dest file /var/log/mosquitto/mosquitto.log

include_dir /etc/mosquitto/conf.d


# adding
allow_anonymous false
password_file /etc/mosquitto/pwfile
acl_file /etc/mosquitto/aclfile


cafile /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_logger


config = 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 = False


class 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_callback
        self.client.on_publish = self.on_publish_callback
        self.client.on_connect = self.connect_callback
        self.conn_flag = False

    def 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

    @staticmethod
    def connect_callback(client, userdata, flags, rc):
        assert rc == 0
        global conn_flag
        conn_flag = True
        pub_logger.info("\nMQTT broker connected successfully,\n"
                        "address: {},\n"
                        "   port: {}\n".format(client._host, client._port))

    @staticmethod
    def disconnect_callback(client, userdata, rc):
        # rc == 0, 代表断开连接成功
        assert rc == 0
        pub_logger.info("\nDisconnect from MQTT broker,\n"
                        "address: {},\n"
                        "   port: {}\n".format(client._host, client._port))

    @staticmethod
    def on_publish_callback(client, userdata, mid):
        topic, msg = userdata
        pub_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 keepalive
        self.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 e


if __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 ssl


config = 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 = False


class 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_callback

        self.client.on_subscribe = self.subscribe_callback

    def 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

    @staticmethod
    def connect_callback(client, userdata, flags, rc):
        # rc == 0 为成功，否则就失败
        assert rc == 0
        global conn_flag
        conn_flag = True
        sub_logger.info("\nMQTT broker connected successfully,\n"
                        "address: {},\n"
                        "   port: {}\n".format(client._host, client._port))

    @staticmethod
    def disconnect_callback(client, userdata, rc):
        # rc == 0 为成功，否则就失败
        assert rc == 0
        sub_logger.info("\nDisconnect from MQTT broker,\n"
                        "address: {},\n"
                        "   port: {}\n".format(client._host, client._port))

    @staticmethod
    def subscribe_callback(client, userdata, mid, granted_qos):
        sub_logger.info("Subscribed successfully.\n")

    @staticmethod
    def on_message_callback(client, userdata, message):
        msg_bytes = message.payload
        msg_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 keepalive
        self.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:
        """
        # 连接broker
        self.connect_mqtt_broker()
        # 订阅的 topic
        self.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.120
  port: 1883
  keepalive: 30
# 订阅者
subscriber:
  # 证书相关参数
  certfile: "cert/ca.crt"
  keyfile: "cert/ca.key"
  cert_pass: "123456"
  # 客户端相关参数
  username: "sub"
  password: "123456"
  topic: "test"
  log_name: mqtt subscriber
  log_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 publisher
  log_file_name: pub.log