Redis 6.0 ACL基于Bitmap实现

2020-03-14

Redis 6.0在4月30日就要和大家正式见面了,现在redis.io上已经提供了RC版本。在之前的博客中,已经介绍过权限控制新功能的一些用法,主要来源于作者Antirez在Redis Day上的一些演示。Antirez在最后提到,ACL的主要实现是基于Bitmap,因此对性能影响是可以忽略不计的。当时大致猜想了一下实现的思路,那么现在离发布已经很近了,作者也对ACL Logging进行了一些补充,不妨一起来看一下。

user结构

server.h中定义了对应的user结构保存用户的ACL信息,包括:

  • 用户名
  • flag,主要是一些特殊状态,例如用户的启用与禁用、整体控制(所有命令可用与否、所有键可访问与否)、免密码等
  • 可用命令(allowed_commands),一个长整型数。每一位代表命令,如果用户允许使用这个命令则置位1
  • 可用子命令(allowed_subcommands),一个指针数组,值也为指针,数组与可用命令一一对应,值为一个SDS数组,SDS数组中存放的是这个命令可用的子命令
  • 用户密码
  • 可用的key patterns。如果这个字段为NULL,用户将不能使用任何Key,除非flag中指明特殊状态如ALLKEYS
typedef struct user {
    sds name;
    uint64_t flags;
    uint64_t allowed_commands[USER_COMMAND_BITS_COUNT/64];
    sds **allowed_subcommands;
    list *passwords;
    list *patterns;
} user;

补充一下一些新鲜的字段描述,allowed_commands实际上是一个(默认)长度为1024的位图,它的index对应各个命令的ID,在历史版本中命令结构redisCommand是通过名字(name)来查找的,id为这个版本中新增的属性,专门用于ACL功能。

struct redisCommand {
    ...
    int id;
};

user这个结构对应的是client结构的”user”字段,熟悉Redis的同学应该对client也有所了解,就不再赘述了。

ACL操作选读

ACL的命令很多,总体而言都是围绕着user对象展开的,因此从中挑选了几个函数来看一下具体是如何操作user对象。

一个需要铺垫的通用方法就是ACLGetUserCommandBit,ACL操作中都会涉及到获取用户的命令位图,ACLGetUserCommandBit()接收一个user结构和命令ID,根据ID定位出命令在allowed_commands中的位置,通过位运算返回用户是否有该命令权限

int ACLGetUserCommandBit(user *u, unsigned long id) {
    uint64_t word, bit;
    if (ACLGetCommandBitCoordinates(id,&word,&bit) == C_ERR) return 0;
    return (u->allowed_commands[word] & bit) != 0;
}

当用户进行Redis操作时,例如set操作,操作的命令会保存在client结构的*cmd字段中,*cmd字段就是一个redisCommand结构的指针,redisCommand结构包含了命令的id,因此在使用时通过ACLGetUserCommandBit(u, cmd->id)传入。

创建用户

创建用户分为两步,首先需要创建一个user,通过调用ACLCreateUser(const char *name, size_t namelen)实现,返回的是一个user对象的指针。在创建时,会在server.h定义的Users中查找是否有同名用户,也是本次功能新增的,因为旧版本中只有”default”用户。此时这个用户拥有名称,flag被初始化为禁用用户,其余的属性均为Null或空list等。

然后,通过调用ACLSetUser(user *u, const char *op, ssize_t oplen),调整传入用户u的对应属性,调整内容放在名为op操作的参数中。这个函数非常长,主要是针对各种不同的“操作” switch case处理,节选部分如下:

int ACLSetUser(user *u, const char *op, ssize_t oplen) {
    if (oplen == -1) oplen = strlen(op);
    /* Part1 - 处理用户状态(flag)操作 */
    // 控制用户启用状态
    if (!strcasecmp(op,"on")) {
        u->flags |= USER_FLAG_ENABLED;
        u->flags &= ~USER_FLAG_DISABLED;
    } else if (!strcasecmp(op,"off")) {
        u->flags |= USER_FLAG_DISABLED;
        u->flags &= ~USER_FLAG_ENABLED;
    // 控制全局键、命令等可用与否
    } else if (!strcasecmp(op,"allkeys") ||
               !strcasecmp(op,"~*"))
    {
        u->flags |= USER_FLAG_ALLKEYS;
        listEmpty(u->patterns);
    }
    ...


    /* Part2 - 操作用户密码增删改查 */
    // > 和 < 等控制密码的改动删除等
    else if (op[0] == '>' || op[0] == '#') {
        sds newpass;
        if (op[0] == '>') {
            newpass = ACLHashPassword((unsigned char*)op+1,oplen-1);
        }


    /* Part3 - 操作用户可用命令的范围 */
    else if (op[0] == '+' && op[1] != '@') {
        if (strchr(op,'|') == NULL) {
            if (ACLLookupCommand(op+1) == NULL) {
                errno = ENOENT;
                return C_ERR;
            }
            unsigned long id = ACLGetCommandID(op+1);
            // 根据传入的id参数设置对应allowed_commands位图的值
            ACLSetUserCommandBit(u,id,1);
            // 新调整的命令的子命令数组会被重置
            ACLResetSubcommandsForCommand(u,id);
        }
    }

补充一下具体调用例子,其实Redis的默认用户就是按照这套流程创建的:初始化名为“default”的空白无权限用户,然后为这个用户设置上所有权限:

DefaultUser = ACLCreateUser("default",7);
ACLSetUser(DefaultUser,"+@all",-1);
ACLSetUser(DefaultUser,"~*",-1);
ACLSetUser(DefaultUser,"on",-1);
ACLSetUser(DefaultUser,"nopass",-1);

拦截不可用命令/键

命令/键拦截操作非常简单:

  • 判断命令/键是否可用
    • 如果不可用,ACL Log处理以及返回错误

ACL判断

我们先看一下“不可用”的判断逻辑,然后再回到命令执行流程中看判断方法的调用。

判断函数同样非常长,展示完后会进行总结:

int ACLCheckCommandPerm(client *c, int *keyidxptr) {
    user *u = c->user;
    uint64_t id = c->cmd->id;
    // 命令相关的全局flag的检查,若满足则跳过后续部分
    if (!(u->flags & USER_FLAG_ALLCOMMANDS) &&
        c->cmd->proc != authCommand)
    {
        // 即使当前命令没有在allowed_commands中,还要检查子命令是否可用
        // 以免出现仅开放了部分子命令权限的情况
        if (ACLGetUserCommandBit(u,id) == 0) {
            ...
            // 遍历子命令
            long subid = 0;
            while (1) {
                if (u->allowed_subcommands[id][subid] == NULL)
                    return ACL_DENIED_CMD;
                if (!strcasecmp(c->argv[1]->ptr,
                                u->allowed_subcommands[id][subid]))
                    break; // 子命令可用,跳出循环
                subid++;
            }
        }
    }

    // 键相关的全局flag检查,若满足则跳过后续部分
    if (!(c->user->flags & USER_FLAG_ALLKEYS) &&
        (c->cmd->getkeys_proc || c->cmd->firstkey))
    {
        int numkeys;
        // 先拿到当前要进行操作的Key
        int *keyidx = getKeysFromCommand(c->cmd,c->argv,c->argc,&numkeys);
        for (int j = 0; j < numkeys; j++) {
            listIter li;
            listNode *ln;
            listRewind(u->patterns,&li);

            // 检查当前user所有的关于Key的匹配Pattern
            // 如果有任意命中则跳出,否则判定不可用
            int match = 0;
            while((ln = listNext(&li))) {
                sds pattern = listNodeValue(ln);
                size_t plen = sdslen(pattern);
                int idx = keyidx[j];
                if (stringmatchlen(pattern,plen,c->argv[idx]->ptr,
                                   sdslen(c->argv[idx]->ptr),0))
                {
                    match = 1;
                    break;
                }
            }
            if (!match) {
                if (keyidxptr) *keyidxptr = keyidx[j];
                getKeysFreeResult(keyidx);
                return ACL_DENIED_KEY;
            }
        }
        getKeysFreeResult(keyidx);
    }
    return ACL_OK;
}

那么为了方便喜欢跳过代码的同学看结论:

  • ACL限制围绕user的各个字段进行
  • 全局的flag优先级最高,例如设置为所有键可用,所有命令可用,会跳过后续的可用命令遍历和可用键Pattern匹配
  • 即使在allowed_commands位图中没有被置位,命令也可能可用,因为它是个子命令,而且命令只开放了部分子命令的使用权限
  • 键通过遍历所有定义了的Pattern检查,如果有匹配上说明可用
  • 先判断操作是否可用,再判断键(包括全局flag也在操作之后)是否可用,两种判断分别对应不同返回整数值:ACL_DENIED_CMDACL_DENIED_KEY

命令执行流程中的调用

判断逻辑之后到何时调用这套判断。我们先来复习一下Redis如何执行命令:

  • 用户操作
  • 客户端RESP协议(Redis 6.0中有RESP3新协议记得关注)压缩发送给服务端
  • 服务端解读消息,存放至client对象的对应字段中,例如argcargv等存放命令和参数等内容
  • 执行前检查(各种执行条件)
  • 执行命令
  • 执行后处理(慢查询日志、AOF等)

目前执行命令的方法是在server.c中的processCommand(client *c),传入client对象,执行,返回执行成功与否。我们节选其中关于ACL的部分如下:

int processCommand(client *c) {
    ...
    int acl_keypos;
    int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
    if (acl_retval != ACL_OK) {
        addACLLogEntry(c,acl_retval,acl_keypos,NULL);
        flagTransaction(c);
        if (acl_retval == ACL_DENIED_CMD)
            addReplyErrorFormat(c,
                "-NOPERM this user has no permissions to run "
                "the '%s' command or its subcommand", c->cmd->name);
        else
            addReplyErrorFormat(c,
                "-NOPERM this user has no permissions to access "
                "one of the keys used as arguments");
        return C_OK;
    }
    ...

在命令解析之后,真正执行之前,通过调用ACLCheckCommandPerm获取判断结果,如果判定不通过,进行以下操作:

  • 记录ACL不通过的日志,这个是作者在RC1之后新增的功能,还在Twitch上进行了直播开发,有兴趣的同学可以在Youtube上看到录播
  • 如果当前处于事务(MULTI)过程中,将client的flag置为CLIENT_DIRTY_EXEC
  • 根据命令还是键不可用,返回给客户端不同的信息

因此这次ACL功能影响的是执行命令前后的操作。

其他功能对ACL的调用

通过搜索可以发现一共有3处调用了ACLCheckCommandPerm方法:

/home/duck/study/redis/src/multi.c:
  179  
  180          int acl_keypos;
  181:         int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
  182          if (acl_retval != ACL_OK) {
  183              addACLLogEntry(c,acl_retval,acl_keypos,NULL);

/home/duck/study/redis/src/scripting.c:
  608      /* Check the ACLs. */
  609      int acl_keypos;
  610:     int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
  611      if (acl_retval != ACL_OK) {
  612          addACLLogEntry(c,acl_retval,acl_keypos,NULL);

/home/duck/study/redis/src/server.c:
 3394       * ACLs. */
 3395      int acl_keypos;
 3396:     int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
 3397      if (acl_retval != ACL_OK) {
 3398          addACLLogEntry(c,acl_retval,acl_keypos,NULL);

形式都是大同小异,了解一下即可。总结一下需要判定ACL的位置:

  • 正常命令执行流程中
  • MULTI事务执行过程中
  • Lua脚本

总结

补充一张图来描述新增的ACL功能相关的结构(点击放大查看):

图中部分的表达可能与实际的数据结构有所差异,主要原因是代码理解和C语言的语法掌握不到位所致。

阅读代码的过程中留意到,对命令的限制是通过Bitmap来实现的,而对Key的限制是通过特定Pattern来实现的。当对Key的限制Pattern数量特别多时,是否会因为匹配Pattern而对性能造成影响,例如超多次的stringmatchlen()执行。当然这一块内容似乎确实没有想到什么提升非常大的判断方式,后续也会继续关注ACL的相关改进。