很多物连网设备并没有做MQTT权限校验，这样会有极大的安全隐患。
1 禁止匿名用户登录
sudo vim /etc/emqx/emqx.conf
allow_anonymous = false
restart
sudo emqx restart
配置之后，打开Dashboard面板测试连接
http://ip:18083/#/websocket
连接的时候没有配置用户名会发现无法连接。
2 验证MQTT CONNECT报文信息
EMQ X支持很多种验证方式，其中一种简单灵活的验证方式为HTTP接口验证。这里简单说一下MQTT协议的连接过程，MQTT基于TCP协议，在建立好一个TCP连接之后，发送的第一个报文必须是CONNECT报文，该报文中有ClientID、UserName、Password三个数据信息。MQTT服务端协议的实现需要根据自己的需要进行校验。当检验通过，MQTT服务器会返回成功的报文。当CONNECT完成，就可以进行订阅发布消息。
为了开启EMQ X的HTTP认证鉴权，我们需要在Dashboard上开启emqx_auth_http功能。
同时，我们查阅EMQ X已有的配置文件信息：/etc/emqx/plugins/emqx_auth_http.conf
auth.http.auth_req = http://127.0.0.1:8991/mqtt/auth
## Value: post | get | put
auth.http.auth_req.method = post
## Value: Params
auth.http.auth_req.params = clientid=%c,username=%u,password=%P
可以看到，默认使用了8991的端口以及POST请求。现在，我们写一个Flask程序进行测试。
from flask import Flask, request, Response
app = Flask(__name__)
@app.route('/mqtt/auth', methods=['POST'])
def test():
print(request.json)
print(request.args)
print(request.form)
if request.form.get('username') != 'zhangsan':
return Response(status=403)
return Response(status=200)
app.run(host='0.0.0.0', port=8991, debug=True)
EMQ X会提交ClientID、UserName、Password信息到我们的接口，我们可以根据自己的业务需求，动态扩展验证需求。返回200状态码代表成功。
3 控制MQTT PUBLISH和SUBSCRIBE控制报文
在CONNECT控制报文完成之后，通常用户需要订阅或者发布，在这个报文上，会携带Topic。开发者可以校验这些信息，进行发布订阅的权限控制。在EMQ X上，提供了一种HTTP接口的方式进行控制。
查看文件：/etc/emqx/plugins/emqx_auth_http.conf
##--------------------------------------------------------------------
## ACL request.
##
## Variables:
## - %A: 1 | 2, 1 = sub, 2 = pub
## - %u: username
## - %c: clientid
## - %a: ipaddress
## - %r: protocol
## - %m: mountpoint
## - %t: topic
##
## Value: URL
auth.http.acl_req = http://127.0.0.1:8991/mqtt/acl
## Value: post | get | put
auth.http.acl_req.method = get
## Value: Params
auth.http.acl_req.params = access=%A,username=%u,clientid=%c,ipaddr=%a,topic=%t,mountpoint=%m
我们还是跟之前一样，用Flask写一个API进行测试
from flask import Flask, request, Response
app = Flask(__name__)
@app.route('/mqtt/auth', methods=['POST'])
def test():
print(request.json)
print(request.args)
print(request.form)
# 对用户登录的信息进行校验
if request.form.get('username') != 'zhangsan':
return Response(status=403)
return Response(status=200)
@app.route('/mqtt/acl', methods=['GET'])
def acl():
print(request.args)
# 订阅access是1
access = request.args.get('access')
return Response(status=200)
# if int(access) == 1:
# print('返回成功')
# return Response(status=200)
# else:
# return Response(status=403)
app.run(host='0.0.0.0', port=8991, debug=True)
同时，修改配置文件/etc/emqx/emqx.conf，使得当ACL没有匹配的我们直接拒绝
## Value: allow | deny
acl_nomatch = deny
当我们取消acl函数的注释代码，我们会发现一个现象，发布消息不会成功。但是Dashboard显示成功发送。
同样的，如果我么在access==1也就是订阅动作的时候，返回非200状态码，那么Dashboard会显示失败，重复操作，不会触发Flask的接口请求。
网友评论