EMQ X：发布订阅ACL

简介

发布订阅 ACL 指对 发布 (PUBLISH)/订阅 (SUBSCRIBE) 操作的 权限控制。例如拒绝用户名为 Anna 向 open/elsa/door 发布消息。

EMQ X 支持通过客户端发布订阅 ACL 进行客户端权限的管理，本章节介绍了 EMQ X 支持的发布订阅 ACL 以及对应插件的配置方法。

ACL 插件

EMQ X 支持使用配置文件、外部主流数据库和自定义 HTTP API 作为 ACL 数据源。

连接数据源、进行访问控制功能是通过插件实现的，使用前需要启用相应的插件。

客户端订阅主题、发布消息时插件通过检查目标主题（Topic）是否在指定数据源允许/禁止列表内来实现对客户端的发布、订阅权限管理。

配置文件/内置数据源

• 内置数据库 认证/访问控制

使用配置文件提供认证数据源，适用于变动较小的 ACL 管理。

外部数据库

• MySQL 认证/访问控制
• PostgreSQL 认证/访问控制
• Redis 认证/访问控制
• MongoDB 认证/访问控制
• LDAP 认证/访问控制

外部数据库可以存储大量数据、动态管理 ACL，方便与外部设备管理系统集成。

其他

• HTTP 认证/访问控制

HTTP ACL 能够实现复杂的 ACL 管理。

ACL 功能包含在认证鉴权插件中，更改插件配置后需要重启插件才能生效

规则详解

ACL 是允许与拒绝条件的集合，EMQ X 中使用以下元素描述 ACL 规则：

## Allow-Deny Who Pub-Sub Topic

"允许(Allow) / 拒绝(Deny)"  "谁(Who)"  "订阅(Subscribe) / 发布(Publish)" "主题列表(Topics)"

同时具有多条 ACL 规则时，EMQ X 将按照规则排序进行合并，以 ACL 文件中的默认 ACL 为例，ACL 文件中配置了默认的 ACL 规则，规则从下至上加载：

1. 第一条规则允许客户端发布订阅所有主题
2. 第二条规则禁止全部客户端订阅 $SYS/# 与 # 主题
3. 第三条规则允许 ip 地址为 127.0.0.1 的客户端发布/订阅 $SYS/# 与 # 主题，为第二条开了特例
4. 第四条规则允许用户名为 dashboard 的客户端订阅 $SYS/# 主题，为第二条开了特例

{allow, {user, "dashboard"}, subscribe, ["$SYS/#"]}.

{allow, {ipaddr, "127.0.0.1"}, pubsub, ["$SYS/#", "#"]}.

{deny, all, subscribe, ["$SYS/#", {eq, "#"}]}.

{allow, all}.

授权结果

任何一次 ACL 授权最终都会返回一个结果：

• 允许：经过检查允许客户端进行操作
• 禁止：经过检查禁止客户端操作
• 忽略（ignore）：未查找到 ACL 权限信息，无法显式判断结果是允许还是禁止，交由下一 ACL 插件或默认 ACL 规则来判断

全局配置

默认配置中 ACL 是开放授权的，即授权结果为忽略（ignore）*时*允许客户端通过授权。

通过 etc/emqx.conf 中的 ACL 配置可以更改该属性：

vi /etc/emqx/emqx.conf

#从allow修改为deny
## Allow or deny if no ACL rules matched.
##
## Value: allow | deny
acl_nomatch = deny

配置默认 ACL 文件，使用文件定义默认 ACL 规则：

## Default ACL File.
##
## Value: File Name
acl_file = /etc/emqx/acl.conf

配置 ACL 授权结果为禁止的响应动作，为 disconnect 时将断开设备：

## The action when acl check reject current operation
##
## Value: ignore | disconnect
## Default: ignore
acl_deny_action = ignore

在 MQTT v3.1 和 v3.1.1 协议中，发布操作被拒绝后服务器无任何报文错误返回，这是协议设计的一个缺陷。但在 MQTT v5.0 协议上已经支持应答一个相应的错误报文。

超级用户（superuser）

客户端可拥有“超级用户”身份，超级用户拥有最高权限不受 ACL 限制。

1. 认证鉴权插件启用超级用户功能后，发布订阅时 EMQ X 将优先检查客户端超级用户身份
2. 客户端为超级用户时，通过授权并跳过后续 ACL 检查

ACL 缓存

ACL 缓存允许客户端在命中某条 ACL 规则后，便将其缓存至内存中，以便下次直接使用，客户端发布、订阅频率较高的情况下开启 ACL 缓存可以提高 ACL 检查性能。

在 emqx.conf 可以配置 ACL 缓存大小与缓存时间：

## 是否启用
enable_acl_cache = on

## 单个客户端最大缓存规则数量
acl_cache_max_size = 32

## 缓存失效时间，超时后缓存将被清除
acl_cache_ttl = 1m

在更新 ACL 规则后，某些客户端由于已经存在缓存，则无法立即生效。若要立即生效，则需手动清除所有的 ACL 缓存：

清除缓存

DELETE /api/v4/acl-cache

清除集群中所有的 ACL 缓存

Query String Parameters: 无

Success Response Body (JSON):

Name	Type	Description
code	Integer	0
message	String	仅在发生错误时返回，用于提供更详细的错误信息

examples:

$ curl -i --basic -u admin:public -X DELETE "http://localhost:8081/api/v4/acl-cache"

{"code":0}

DELETE /api/v4/node/{node}/acl-cache

清除指定节点的 ACL 缓存

Query String Parameters: 无

Success Response Body (JSON):

Name	Type	Description
code	Integer	0
message	String	仅在发生错误时返回，用于提供更详细的错误信息

Examples:

$ curl -i --basic -u admin:public -X DELETE "http://localhost:8081/api/v4/node/emqx@127.0.0.1/acl-cache"

{"code":0}

ACL 鉴权链

当同时启用多个 ACL 插件时，EMQ X 将按照插件开启先后顺序进行链式鉴权：

• 一通过授权，终止链并允许客户端通过验证

• 一旦授权失败，终止链并禁止客户端通过验证

• 直到最后一个 ACL 插件仍未通过，根据

  默认授权

  配置判定

  • 默认授权为允许时，允许客户端通过验证
  • 默认授权为禁止时，禁止客户端通过验证

同时只启用一个 ACL 插件可以提高客户端 ACL 检查性能。

内置ACL说明

内置 ACL 通过文件设置规则，使用上足够简单轻量，适用于规则数量可预测、无变动需求或变动较小的项目。

内置 ACL 优先级最低，可以被 ACL 插件覆盖，如需禁用全部注释即可。规则文件更改后需重启 EMQ X 以应用生效。

在 acl.conf 修改完成后，并不会自动加载至 EMQ X 系统。需要手动执行：

emqx_ctl acl reload

acl.conf 中应只包含一些简单而通用的规则，使其成为系统基础的 ACL 原则。如果需要支持复杂、大量的 ACL 内容，需要使用认证插件。

下面介绍Http ACL的使用。

HTTP ACL

http acl插件配置在/etc/emqx/plugins/emqx_auth_http.conf配置文件中。

ACL授权原理

EMQ X 在设备发布、订阅事件中使用当前客户端相关信息作为参数，向用户自定义的认证服务发起请求权限，通过返回的 HTTP 响应状态码 (HTTP statusCode) 来处理 ACL 授权请求。

• 无权限：API 返回 4xx 状态码
• 授权成功：API 返回 200 状态码
• 忽略授权：API 返回 200 状态码且消息体 ignore

实战

修改配置文件

修改/etc/emqx/plugins/emqx_auth_http.conf

#其余未修改选项省略
auth.http.auth_req.url = http://127.0.0.1:8991/mqtt/auth
auth.http.super_req.url = http://127.0.0.1:8991/mqtt/superuser
auth.http.acl_req.url = http://127.0.0.1:8991/mqtt/acl

授权接口开发

@RestController
@RequestMapping("/mqtt")
public class AuthController {

    Logger logger = LoggerFactory.getLogger(AuthController.class);

    Map<String, String> userMap = new HashMap<>();

    @PostConstruct
    public void init(){
        userMap.put("user1","user1");
        userMap.put("user2","user2");
        userMap.put("admin","admin");
    }


    @PostMapping("/auth")
    public ResponseEntity<HttpStatus> auth(String username, String clientid, String password){
        logger.info("开始进入auth方法");
        logger.info("username:{}; clientid:{}; password:{}", username, clientid, password);
        String passwd = userMap.get(username);
        if(!StringUtils.hasLength(passwd)){
            return new ResponseEntity<>(HttpStatus.UNAUTHORIZED);
        }
        return !password.equals(passwd) ? new ResponseEntity<>(HttpStatus.UNAUTHORIZED) : new ResponseEntity<>(HttpStatus.OK);
    }

    @PostMapping("/superuser")
    public ResponseEntity<HttpStatus> superuser(String clientid, String username){
        logger.info("开始进入superuser方法");
        logger.info("username:{}; clientid:{}", username, clientid);
        if("admin".equals(clientid) || username.equals("admin")){
            return new ResponseEntity<>(HttpStatus.OK);
        }
        return new ResponseEntity<>(HttpStatus.UNAUTHORIZED);
    }

    /**
     * 在这个方法中我们设置了：
     * 只有用户名为 user1 的客户端能够去订阅 testtopic/# 的权限其他客户端都不可以
     * 只有用户名为 user2 的客户端能够向 testtopic/123 发布消息的权限其他都没有
     * @param access (1 - subscribe, 2 - publish)
     * @author wen.jie
     * @date 2021/7/28 10:01
     */
    @PostMapping("/acl")
    public ResponseEntity<HttpStatus> acl(int access,String username,String clientid,String ipaddr,String topic,String mountpoint){
        logger.info("EMQX发起客户端操作授权查询请求,access={},username={},clientid={},ipaddr= {},topic={},mountpoint={}", access,username,clientid,ipaddr,topic,mountpoint);
        if(username.equals("user1") && topic.equals("testtopic/#") && access == 1){
            logger.info("客户端{}有权限订阅{}",username,topic);
            return new ResponseEntity<>(HttpStatus.OK);
        }
        if(username.equals("user2") && topic.equals("testtopic/123") && access == 2){
            logger.info("客户端{}有权限向{}发布消息",username,topic);
            return new ResponseEntity<>(null, HttpStatus.OK);
        }
        logger.info("客户端{},username={},没有权限对主题{}进行{}操作",clientid,username,topic,access==1?"订阅":"发布");
        return new ResponseEntity<>(HttpStatus.UNAUTHORIZED);//无权限
    }
}

将该项目部署到linux中，并测试

HTTP ACL测试

测试admin登录：

添加订阅：

测试发送消息：

user1有权订阅testtopic/#主题：

测试user1发送消息但是其他人并没有收到消息

测试user2发送消息和订阅主题