EMQ X 认证鉴权以及ACL鉴权（HTTP方式）

EMQ X document link

很多物连网设备并没有做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的接口请求。