Merge branch 'unstable' of https://github.com/antirez/redis into unstable

redis.conf

@@ -493,20 +493,39 @@ replica-priority 100
 
 ################################## SECURITY ###################################
 
-# Require clients to issue AUTH <PASSWORD> before processing any other
-# commands.  This might be useful in environments in which you do not trust
-# others with access to the host running redis-server.
-#
-# This should stay commented out for backward compatibility and because most
-# people do not need auth (e.g. they run their own servers).
-#
 # Warning: since Redis is pretty fast an outside user can try up to
-# 150k passwords per second against a good box. This means that you should
+# 1 million passwords per second against a modern box. This means that you
-# use a very strong password otherwise it will be very easy to break.
+# should use very strong passwords, otherwise they will be very easy to break.
+# Note that because the password is really a shared secret between the client
+# and the server, and should not be memorized by any human, the password
+# can be easily a long string from /dev/urandom or whatever, so by using a
+# long and unguessable password no brute force attack will be possible.
+
+# Instead of configuring users here in this file, it is possible to use
+# a stand-alone file just listing users. The two methods cannot be mixed:
+# if you configure users here and at the same time you activate the exteranl
+# ACL file, the server will refuse to start.
+#
+# The format of the external ACL user file is exactly the same as the
+# format that is used inside redis.conf to describe users.
+#
+# aclfile /etc/redis/users.acl
+
+# IMPORTANT NOTE: starting with Redis 6 "requirepass" is just a compatiblity
+# layer on top of the new ACL system. The option effect will be just setting
+# the password for the default user. Clients will still authenticate using
+# AUTH <password> as usually, or more explicitly with AUTH default <password>
+# if they follow the new protocol: both will work.
 #
 # requirepass foobared
 
-# Command renaming.
+# Command renaming (DEPRECATED).
+#
+# ------------------------------------------------------------------------
+# WARNING: avoid using this option if possible. Instead use ACLs to remove
+# commands from the default user, and put them only in some admin user you
+# create for administrative purposes.
+# ------------------------------------------------------------------------
 #
 # It is possible to change the name of dangerous commands in a shared
 # environment. For instance the CONFIG command may be renamed into something

src/acl.c

@@ -34,10 +34,19 @@
  * ==========================================================================*/
 
 rax *Users; /* Table mapping usernames to user structures. */
-user *DefaultUser;   /* Global reference to the default user.
+
-                        Every new connection is associated to it, if no
+user *DefaultUser;  /* Global reference to the default user.
-                        AUTH or HELLO is used to authenticate with a
+                       Every new connection is associated to it, if no
-                        different user. */
+                       AUTH or HELLO is used to authenticate with a
+                       different user. */
+
+list *UsersToLoad;  /* This is a list of users found in the configuration file
+                       that we'll need to load in the final stage of Redis
+                       initialization, after all the modules are already
+                       loaded. Every list element is a NULL terminated
+                       array of SDS pointers: the first is the user name,
+                       all the remaining pointers are ACL rules in the same
+                       format as ACLSetUser(). */
 
 struct ACLCategoryItem {
     const char *name;
@@ -64,9 +73,24 @@ struct ACLCategoryItem {
     {"connection", CMD_CATEGORY_CONNECTION},
     {"transaction", CMD_CATEGORY_TRANSACTION},
     {"scripting", CMD_CATEGORY_SCRIPTING},
-    {"",0} /* Terminator. */
+    {NULL,0} /* Terminator. */
 };
 
+struct ACLUserFlag {
+    const char *name;
+    uint64_t flag;
+} ACLUserFlags[] = {
+    {"on", USER_FLAG_ENABLED},
+    {"off", USER_FLAG_DISABLED},
+    {"allkeys", USER_FLAG_ALLKEYS},
+    {"allcommands", USER_FLAG_ALLCOMMANDS},
+    {"nopass", USER_FLAG_NOPASS},
+    {NULL,0} /* Terminator. */
+};
+
+void ACLResetSubcommandsForCommand(user *u, unsigned long id);
+void ACLResetSubcommands(user *u);
+
 /* =============================================================================
  * Helper functions for the rest of the ACL implementation
  * ==========================================================================*/
@@ -148,7 +172,7 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     if (raxFind(Users,(unsigned char*)name,namelen) != raxNotFound) return NULL;
     user *u = zmalloc(sizeof(*u), MALLOC_LOCAL);
     u->name = sdsnewlen(name,namelen);
-    u->flags = 0;
+    u->flags = USER_FLAG_DISABLED;
     u->allowed_subcommands = NULL;
     u->passwords = listCreate();
     u->patterns = listCreate();
@@ -161,6 +185,16 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     return u;
 }
 
+/* Release the memory used by the user structure. Note that this function
+ * will not remove the user from the Users global radix tree. */
+void ACLFreeUser(user *u) {
+    sdsfree(u->name);
+    listRelease(u->passwords);
+    listRelease(u->patterns);
+    ACLResetSubcommands(u);
+    zfree(u);
+}
+
 /* Given a command ID, this function set by reference 'word' and 'bit'
  * so that user->allowed_commands[word] will address the right word
  * where the corresponding bit for the provided ID is stored, and
@@ -222,12 +256,194 @@ int ACLSetUserCommandBitsForCategory(user *u, const char *category, int value) {
     dictEntry *de;
     while ((de = dictNext(di)) != NULL) {
         struct redisCommand *cmd = dictGetVal(de);
-        if (cmd->flags & cflag) ACLSetUserCommandBit(u,cmd->id,value);
+        if (cmd->flags & cflag) {
+            ACLSetUserCommandBit(u,cmd->id,value);
+            ACLResetSubcommandsForCommand(u,cmd->id);
+        }
     }
     dictReleaseIterator(di);
     return C_OK;
 }
 
+/* Return the number of commands allowed (on) and denied (off) for the user 'u'
+ * in the subset of commands flagged with the specified category name.
+ * If the categoty name is not valid, C_ERR is returend, otherwise C_OK is
+ * returned and on and off are populated by reference. */
+int ACLCountCategoryBitsForUser(user *u, unsigned long *on, unsigned long *off,
+                                const char *category)
+{
+    uint64_t cflag = ACLGetCommandCategoryFlagByName(category);
+    if (!cflag) return C_ERR;
+
+    *on = *off = 0;
+    dictIterator *di = dictGetIterator(server.orig_commands);
+    dictEntry *de;
+    while ((de = dictNext(di)) != NULL) {
+        struct redisCommand *cmd = dictGetVal(de);
+        if (cmd->flags & cflag) {
+            if (ACLGetUserCommandBit(u,cmd->id))
+                (*on)++;
+            else
+                (*off)++;
+        }
+    }
+    dictReleaseIterator(di);
+    return C_OK;
+}
+
+/* This function returns an SDS string representing the specified user ACL
+ * rules related to command execution, in the same format you could set them
+ * back using ACL SETUSER. The function will return just the set of rules needed
+ * to recreate the user commands bitmap, without including other user flags such
+ * as on/off, passwords and so forth. The returned string always starts with
+ * the +@all or -@all rule, depending on the user bitmap, and is followed, if
+ * needed, by the other rules needed to narrow or extend what the user can do. */
+sds ACLDescribeUserCommandRules(user *u) {
+    sds rules = sdsempty();
+    int additive;   /* If true we start from -@all and add, otherwise if
+                       false we start from +@all and remove. */
+
+    /* This code is based on a trick: as we generate the rules, we apply
+     * them to a fake user, so that as we go we still know what are the
+     * bit differences we should try to address by emitting more rules. */
+    user fu = {0};
+    user *fakeuser = &fu;
+
+    /* Here we want to understand if we should start with +@all and remove
+     * the commands corresponding to the bits that are not set in the user
+     * commands bitmap, or the contrary. Note that semantically the two are
+     * different. For instance starting with +@all and subtracting, the user
+     * will be able to execute future commands, while -@all and adding will just
+     * allow the user the run the selected commands and/or categories.
+     * How do we test for that? We use the trick of a reserved command ID bit
+     * that is set only by +@all (and its alias "allcommands"). */
+    if (ACLUserCanExecuteFutureCommands(u)) {
+        additive = 0;
+        rules = sdscat(rules,"+@all ");
+        ACLSetUser(fakeuser,"+@all",-1);
+    } else {
+        additive = 1;
+        rules = sdscat(rules,"-@all ");
+        ACLSetUser(fakeuser,"-@all",-1);
+    }
+
+    /* Try to add or subtract each category one after the other. Often a
+     * single category will not perfectly match the set of commands into
+     * it, so at the end we do a final pass adding/removing the single commands
+     * needed to make the bitmap exactly match. */
+    for (int j = 0; ACLCommandCategories[j].flag != 0; j++) {
+        unsigned long on, off;
+        ACLCountCategoryBitsForUser(u,&on,&off,ACLCommandCategories[j].name);
+        if ((additive && on > off) || (!additive && off > on)) {
+            sds op = sdsnewlen(additive ? "+@" : "-@", 2);
+            op = sdscat(op,ACLCommandCategories[j].name);
+            ACLSetUser(fakeuser,op,-1);
+            rules = sdscatsds(rules,op);
+            rules = sdscatlen(rules," ",1);
+            sdsfree(op);
+        }
+    }
+
+    /* Fix the final ACLs with single commands differences. */
+    dictIterator *di = dictGetIterator(server.orig_commands);
+    dictEntry *de;
+    while ((de = dictNext(di)) != NULL) {
+        struct redisCommand *cmd = dictGetVal(de);
+        int userbit = ACLGetUserCommandBit(u,cmd->id);
+        int fakebit = ACLGetUserCommandBit(fakeuser,cmd->id);
+        if (userbit != fakebit) {
+            rules = sdscatlen(rules, userbit ? "+" : "-", 1);
+            rules = sdscat(rules,cmd->name);
+            rules = sdscatlen(rules," ",1);
+            ACLSetUserCommandBit(fakeuser,cmd->id,userbit);
+        }
+
+        /* Emit the subcommands if there are any. */
+        if (userbit == 0 && u->allowed_subcommands &&
+            u->allowed_subcommands[cmd->id])
+        {
+            for (int j = 0; u->allowed_subcommands[cmd->id][j]; j++) {
+                rules = sdscatlen(rules,"+",1);
+                rules = sdscat(rules,cmd->name);
+                rules = sdscatlen(rules,"|",1);
+                rules = sdscatsds(rules,u->allowed_subcommands[cmd->id][j]);
+                rules = sdscatlen(rules," ",1);
+            }
+        }
+    }
+    dictReleaseIterator(di);
+
+    /* Trim the final useless space. */
+    sdsrange(rules,0,-2);
+
+    /* This is technically not needed, but we want to verify that now the
+     * predicted bitmap is exactly the same as the user bitmap, and abort
+     * otherwise, because aborting is better than a security risk in this
+     * code path. */
+    if (memcmp(fakeuser->allowed_commands,
+                        u->allowed_commands,
+                        sizeof(u->allowed_commands)) != 0)
+    {
+        serverLog(LL_WARNING,
+            "CRITICAL ERROR: User ACLs don't match final bitmap: '%s'",
+            rules);
+        serverPanic("No bitmap match in ACLDescribeUserCommandRules()");
+    }
+    return rules;
+}
+
+/* This is similar to ACLDescribeUserCommandRules(), however instead of
+ * describing just the user command rules, everything is described: user
+ * flags, keys, passwords and finally the command rules obtained via
+ * the ACLDescribeUserCommandRules() function. This is the function we call
+ * when we want to rewrite the configuration files describing ACLs and
+ * in order to show users with ACL LIST. */
+sds ACLDescribeUser(user *u) {
+    sds res = sdsempty();
+
+    /* Flags. */
+    for (int j = 0; ACLUserFlags[j].flag; j++) {
+        /* Skip the allcommands and allkeys flags because they'll be emitted
+         * later as ~* and +@all. */
+        if (ACLUserFlags[j].flag == USER_FLAG_ALLKEYS ||
+            ACLUserFlags[j].flag == USER_FLAG_ALLCOMMANDS) continue;
+        if (u->flags & ACLUserFlags[j].flag) {
+            res = sdscat(res,ACLUserFlags[j].name);
+            res = sdscatlen(res," ",1);
+        }
+    }
+
+    /* Passwords. */
+    listIter li;
+    listNode *ln;
+    listRewind(u->passwords,&li);
+    while((ln = listNext(&li))) {
+        sds thispass = listNodeValue(ln);
+        res = sdscatlen(res,">",1);
+        res = sdscatsds(res,thispass);
+        res = sdscatlen(res," ",1);
+    }
+
+    /* Key patterns. */
+    if (u->flags & USER_FLAG_ALLKEYS) {
+        res = sdscatlen(res,"~* ",3);
+    } else {
+        listRewind(u->patterns,&li);
+        while((ln = listNext(&li))) {
+            sds thispat = listNodeValue(ln);
+            res = sdscatlen(res,"~",1);
+            res = sdscatsds(res,thispat);
+            res = sdscatlen(res," ",1);
+        }
+    }
+
+    /* Command rules. */
+    sds rules = ACLDescribeUserCommandRules(u);
+    res = sdscatsds(res,rules);
+    sdsfree(rules);
+    return res;
+}
+
 /* Get a command from the original command table, that is not affected
  * by the command renaming operations: we base all the ACL work from that
  * table, so that ACLs are valid regardless of command renaming. */
@@ -253,8 +469,13 @@ void ACLResetSubcommandsForCommand(user *u, unsigned long id) {
  * for the user. */
 void ACLResetSubcommands(user *u) {
     if (u->allowed_subcommands == NULL) return;
-    for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++)
+    for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
-        if (u->allowed_subcommands[j]) zfree(u->allowed_subcommands[j]);
+        if (u->allowed_subcommands[j]) {
+            for (int i = 0; u->allowed_subcommands[j][i]; i++)
+                sdsfree(u->allowed_subcommands[j][i]);
+            zfree(u->allowed_subcommands[j]);
+        }
+    }
     zfree(u->allowed_subcommands);
     u->allowed_subcommands = NULL;
 }
@@ -353,12 +574,19 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  *
  * EINVAL: The specified opcode is not understood.
  * ENOENT: The command name or command category provided with + or - is not
- * known. */
+ *         known.
+ * EBUSY:  The subcommand you want to add is about a command that is currently
+ *         fully added.
+ * EEXIST: You are adding a key pattern after "*" was already added. This is
+ *         almost surely an error on the user side.
+ */
 int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     if (oplen == -1) oplen = strlen(op);
     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,"~*"))
@@ -398,6 +626,10 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
         if (ln) listDelNode(u->passwords,ln);
         sdsfree(delpass);
     } else if (op[0] == '~') {
+        if (u->flags & USER_FLAG_ALLKEYS) {
+            errno = EEXIST;
+            return C_ERR;
+        }
         sds newpat = sdsnewlen(op+1,oplen-1);
         listNode *ln = listSearchKey(u->patterns,newpat);
         /* Avoid re-adding the same pattern multiple times. */
@@ -422,6 +654,7 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
             /* Check if the command exists. We can't check the
              * subcommand to see if it is valid. */
             if (ACLLookupCommand(copy) == NULL) {
+                zfree(copy);
                 errno = ENOENT;
                 return C_ERR;
             }
@@ -435,6 +668,15 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
                 return C_ERR;
             }
 
+            /* The command should not be set right now in the command
+             * bitmap, because adding a subcommand of a fully added
+             * command is probably an error on the user side. */
+            if (ACLGetUserCommandBit(u,id) == 1) {
+                zfree(copy);
+                errno = EBUSY;
+                return C_ERR;
+            }
+
             /* Add the subcommand to the list of valid ones. */
             ACLAddAllowedSubcommand(u,id,sub);
 
@@ -469,6 +711,26 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     return C_OK;
 }
 
+/* Return a description of the error that occurred in ACLSetUser() according to
+ * the errno value set by the function on error. */
+char *ACLSetUserStringError(void) {
+    char *errmsg = "Wrong format";
+    if (errno == ENOENT)
+        errmsg = "Unknown command or category name in ACL";
+    else if (errno == EINVAL)
+        errmsg = "Syntax error";
+    else if (errno == EBUSY)
+        errmsg = "Adding a subcommand of a command already fully "
+                 "added is not allowed. Remove the command to start. "
+                 "Example: -DEBUG +DEBUG|DIGEST";
+    else if (errno == EEXIST)
+        errmsg = "Adding a pattern after the * pattern (or the "
+                 "'allkeys' flag) is not valid and does not have any "
+                 "effect. Try 'resetkeys' to start with an empty "
+                 "list of patterns";
+    return errmsg;
+}
+
 /* Return the first password of the default user or NULL.
  * This function is needed for backward compatibility with the old
  * directive "requirepass" when Redis supported a single global
@@ -482,6 +744,7 @@ sds ACLDefaultUserFirstPassword(void) {
 /* Initialization of the ACL subsystem. */
 void ACLInit(void) {
     Users = raxNew();
+    UsersToLoad = listCreate();
     DefaultUser = ACLCreateUser("default",7);
     ACLSetUser(DefaultUser,"+@all",-1);
     ACLSetUser(DefaultUser,"~*",-1);
@@ -503,7 +766,7 @@ int ACLCheckUserCredentials(robj *username, robj *password) {
     }
 
     /* Disabled users can't login. */
-    if ((u->flags & USER_FLAG_ENABLED) == 0) {
+    if (u->flags & USER_FLAG_DISABLED) {
         errno = EINVAL;
         return C_ERR;
     }
@@ -651,6 +914,101 @@ int ACLCheckCommandPerm(client *c) {
     return ACL_OK;
 }
 
+/* =============================================================================
+ * ACL loading / saving functions
+ * ==========================================================================*/
+
+/* Given an argument vector describing a user in the form:
+ *
+ *      user <username> ... ACL rules and flags ...
+ *
+ * this function validates, and if the syntax is valid, appends
+ * the user definition to a list for later loading.
+ *
+ * The rules are tested for validity and if there obvious syntax errors
+ * the function returns C_ERR and does nothing, otherwise C_OK is returned
+ * and the user is appended to the list.
+ *
+ * Note that this function cannot stop in case of commands that are not found
+ * and, in that case, the error will be emitted later, because certain
+ * commands may be defined later once modules are loaded.
+ *
+ * When an error is detected and C_ERR is returned, the function populates
+ * by reference (if not set to NULL) the argc_err argument with the index
+ * of the argv vector that caused the error. */
+int ACLAppendUserForLoading(sds *argv, int argc, int *argc_err) {
+    if (argc < 2 || strcasecmp(argv[0],"user")) {
+        if (argc_err) *argc_err = 0;
+        return C_ERR;
+    }
+
+    /* Try to apply the user rules in a fake user to see if they
+     * are actually valid. */
+    char *funame = "__fakeuser__";
+    user *fakeuser = ACLCreateUser(funame,strlen(funame));
+    serverAssert(fakeuser != NULL);
+    int retval = raxRemove(Users,(unsigned char*) funame,strlen(funame),NULL);
+    serverAssert(retval != 0);
+
+    for (int j = 2; j < argc; j++) {
+        if (ACLSetUser(fakeuser,argv[j],sdslen(argv[j])) == C_ERR) {
+            if (errno != ENOENT) {
+                ACLFreeUser(fakeuser);
+                if (argc_err) *argc_err = j;
+                return C_ERR;
+            }
+        }
+    }
+
+    /* Rules look valid, let's append the user to the list. */
+    sds *copy = zmalloc(sizeof(sds)*argc, MALLOC_LOCAL);
+    for (int j = 1; j < argc; j++) copy[j-1] = sdsdup(argv[j]);
+    copy[argc-1] = NULL;
+    listAddNodeTail(UsersToLoad,copy);
+    ACLFreeUser(fakeuser);
+    return C_OK;
+}
+
+/* This function will load the configured users appended to the server
+ * configuration via ACLAppendUserForLoading(). On loading errors it will
+ * log an error and return C_ERR, otherwise C_OK will be returned. */
+int ACLLoadConfiguredUsers(void) {
+    listIter li;
+    listNode *ln;
+    listRewind(UsersToLoad,&li);
+    while ((ln = listNext(&li)) != NULL) {
+        sds *aclrules = listNodeValue(ln);
+        sds username = aclrules[0];
+        user *u = ACLCreateUser(username,sdslen(username));
+        if (!u) {
+            u = ACLGetUserByName(username,sdslen(username));
+            serverAssert(u != NULL);
+            ACLSetUser(u,"reset",-1);
+        }
+
+        /* Load every rule defined for this user. */
+        for (int j = 1; aclrules[j]; j++) {
+            if (ACLSetUser(u,aclrules[j],sdslen(aclrules[j])) != C_OK) {
+                char *errmsg = ACLSetUserStringError();
+                serverLog(LL_WARNING,"Error loading ACL rule '%s' for "
+                                     "the user named '%s': %s",
+                          aclrules[j],aclrules[0],errmsg);
+                return C_ERR;
+            }
+        }
+
+        /* Having a disabled user in the configuration may be an error,
+         * warn about it without returning any error to the caller. */
+        if (u->flags & USER_FLAG_DISABLED) {
+            serverLog(LL_NOTICE, "The user '%s' is disabled (there is no "
+                                 "'on' modifier in the user description). Make "
+                                 "sure this is not a configuration error.",
+                      aclrules[0]);
+        }
+    }
+    return C_OK;
+}
+
 /* =============================================================================
  * ACL related commands
  * ==========================================================================*/
@@ -671,11 +1029,7 @@ void aclCommand(client *c) {
         serverAssert(u != NULL);
         for (int j = 3; j < c->argc; j++) {
             if (ACLSetUser(u,c->argv[j]->ptr,sdslen(c->argv[j]->ptr)) != C_OK) {
-                char *errmsg = "wrong format";
+                char *errmsg = ACLSetUserStringError();
-                if (errno == ENOENT)
-                    errmsg = "unknown command or category name in ACL";
-                else if (errno == EINVAL)
-                    errmsg = "syntax error";
                 addReplyErrorFormat(c,
                     "Error in ACL SETUSER modifier '%s': %s",
                     (char*)c->argv[j]->ptr, errmsg);
@@ -683,12 +1037,44 @@ void aclCommand(client *c) {
             }
         }
         addReply(c,shared.ok);
-    } else if (!strcasecmp(sub,"whoami")) {
+    } else if (!strcasecmp(sub,"deluser") && c->argc >= 3) {
-        if (c->puser != NULL) {
+        int deleted = 0;
-            addReplyBulkCBuffer(c,c->puser->name,sdslen(c->puser->name));
+        for (int j = 2; j < c->argc; j++) {
-        } else {
+            sds username = c->argv[j]->ptr;
-            addReplyNull(c);
+            if (!strcmp(username,"default")) {
+                addReplyError(c,"The 'default' user cannot be removed");
+                return;
+            }
+            user *u;
+            if (raxRemove(Users,(unsigned char*)username,
+                          sdslen(username),
+                          (void**)&u))
+            {
+                /* When a user is deleted we need to cycle the active
+                 * connections in order to kill all the pending ones that
+                 * are authenticated with such user. */
+                ACLFreeUser(u);
+                listIter li;
+                listNode *ln;
+                listRewind(server.clients,&li);
+                while ((ln = listNext(&li)) != NULL) {
+                    client *c = listNodeValue(ln);
+                    if (c->puser == u) {
+                        /* We'll free the conenction asynchronously, so
+                         * in theory to set a different user is not needed.
+                         * However if there are bugs in Redis, soon or later
+                         * this may result in some security hole: it's much
+                         * more defensive to set the default user and put
+                         * it in non authenticated mode. */
+                        c->puser = DefaultUser;
+                        c->authenticated = 0;
+                        freeClientAsync(c);
+                    }
+                }
+                deleted++;
+            }
         }
+        addReplyLongLong(c,deleted);
     } else if (!strcasecmp(sub,"getuser") && c->argc == 3) {
         user *u = ACLGetUserByName(c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
         if (u == NULL) {
@@ -696,30 +1082,17 @@ void aclCommand(client *c) {
             return;
         }
 
-        addReplyMapLen(c,2);
+        addReplyMapLen(c,4);
 
         /* Flags */
         addReplyBulkCString(c,"flags");
         void *deflen = addReplyDeferredLen(c);
         int numflags = 0;
-        if (u->flags & USER_FLAG_ENABLED) {
+        for (int j = 0; ACLUserFlags[j].flag; j++) {
-            addReplyBulkCString(c,"on");
+            if (u->flags & ACLUserFlags[j].flag) {
-            numflags++;
+                addReplyBulkCString(c,ACLUserFlags[j].name);
-        } else {
+                numflags++;
-            addReplyBulkCString(c,"off");
+            }
-            numflags++;
-        }
-        if (u->flags & USER_FLAG_ALLKEYS) {
-            addReplyBulkCString(c,"allkeys");
-            numflags++;
-        }
-        if (u->flags & USER_FLAG_ALLCOMMANDS) {
-            addReplyBulkCString(c,"allcommands");
-            numflags++;
-        }
-        if (u->flags & USER_FLAG_NOPASS) {
-            addReplyBulkCString(c,"nopass");
-            numflags++;
         }
         setDeferredSetLen(c,deflen,numflags);
 
@@ -733,13 +1106,65 @@ void aclCommand(client *c) {
             sds thispass = listNodeValue(ln);
             addReplyBulkCBuffer(c,thispass,sdslen(thispass));
         }
+
+        /* Commands */
+        addReplyBulkCString(c,"commands");
+        sds cmddescr = ACLDescribeUserCommandRules(u);
+        addReplyBulkSds(c,cmddescr);
+
+        /* Key patterns */
+        addReplyBulkCString(c,"keys");
+        if (u->flags & USER_FLAG_ALLKEYS) {
+            addReplyArrayLen(c,1);
+            addReplyBulkCBuffer(c,"*",1);
+        } else {
+            addReplyArrayLen(c,listLength(u->patterns));
+            listIter li;
+            listNode *ln;
+            listRewind(u->patterns,&li);
+            while((ln = listNext(&li))) {
+                sds thispat = listNodeValue(ln);
+                addReplyBulkCBuffer(c,thispat,sdslen(thispat));
+            }
+        }
+    } else if ((!strcasecmp(sub,"list") || !strcasecmp(sub,"users")) &&
+               c->argc == 2)
+    {
+        int justnames = !strcasecmp(sub,"users");
+        addReplyArrayLen(c,raxSize(Users));
+        raxIterator ri;
+        raxStart(&ri,Users);
+        raxSeek(&ri,"^",NULL,0);
+        while(raxNext(&ri)) {
+            user *u = ri.data;
+            if (justnames) {
+                addReplyBulkCBuffer(c,u->name,sdslen(u->name));
+            } else {
+                /* Return information in the configuration file format. */
+                sds config = sdsnew("user ");
+                config = sdscatsds(config,u->name);
+                config = sdscatlen(config," ",1);
+                sds descr = ACLDescribeUser(u);
+                config = sdscatsds(config,descr);
+                sdsfree(descr);
+                addReplyBulkSds(c,config);
+            }
+        }
+        raxStop(&ri);
+    } else if (!strcasecmp(sub,"whoami")) {
+        if (c->puser != NULL) {
+            addReplyBulkCBuffer(c,c->puser->name,sdslen(c->puser->name));
+        } else {
+            addReplyNull(c);
+        }
     } else if (!strcasecmp(sub,"help")) {
         const char *help[] = {
-"LIST                              -- List all the registered users.",
+"LIST                              -- Show user details in config file format.",
+"USERS                             -- List all the registered usernames.",
 "SETUSER <username> [attribs ...]  -- Create or modify a user.",
-"DELUSER <username>                -- Delete a user.",
 "GETUSER <username>                -- Get the user details.",
-"WHOAMI                            -- Return the current username.",
+"DELUSER <username>                -- Delete a user.",
+"WHOAMI                            -- Return the current connection username.",
 NULL
         };
         addReplyHelp(c,help);

src/config.c

@@ -283,6 +283,9 @@ void loadServerConfigFromString(char *config) {
                 }
                 fclose(logfp);
             }
+        } else if (!strcasecmp(argv[0],"aclfile") && argc == 2) {
+            zfree(server.acl_filename);
+            server.acl_filename = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"always-show-logo") && argc == 2) {
             if ((server.always_show_logo = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
@@ -791,6 +794,16 @@ void loadServerConfigFromString(char *config) {
                     "Allowed values: 'upstart', 'systemd', 'auto', or 'no'";
                 goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"user") && argc >= 2) {
+            int argc_err;
+            if (ACLAppendUserForLoading(argv,argc,&argc_err) == C_ERR) {
+                char buf[1024];
+                char *errmsg = ACLSetUserStringError();
+                snprintf(buf,sizeof(buf),"Error in user declaration '%s': %s",
+                    argv[argc_err],errmsg);
+                err = buf;
+                goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"loadmodule") && argc >= 2) {
             queueLoadModule(argv[1],&argv[2],argc-2);
         } else if (!strcasecmp(argv[0],"sentinel")) {
@@ -1347,6 +1360,7 @@ void configGetCommand(client *c) {
     config_get_string_field("cluster-announce-ip",server.cluster_announce_ip);
     config_get_string_field("unixsocket",server.unixsocket);
     config_get_string_field("logfile",server.logfile);
+    config_get_string_field("aclfile",server.acl_filename);
     config_get_string_field("pidfile",server.pidfile);
     config_get_string_field("slave-announce-ip",server.slave_announce_ip);
     config_get_string_field("replica-announce-ip",server.slave_announce_ip);
@@ -1909,6 +1923,38 @@ void rewriteConfigSaveOption(struct rewriteConfigState *state) {
     rewriteConfigMarkAsProcessed(state,"save");
 }
 
+/* Rewrite the user option. */
+void rewriteConfigUserOption(struct rewriteConfigState *state) {
+    /* If there is a user file defined we just mark this configuration
+     * directive as processed, so that all the lines containing users
+     * inside the config file gets discarded. */
+    if (server.acl_filename[0] != '\0') {
+        rewriteConfigMarkAsProcessed(state,"user");
+        return;
+    }
+
+    /* Otherwise scan the list of users and rewrite every line. Note that
+     * in case the list here is empty, the effect will just be to comment
+     * all the users directive inside the config file. */
+    raxIterator ri;
+    raxStart(&ri,Users);
+    raxSeek(&ri,"^",NULL,0);
+    while(raxNext(&ri)) {
+        user *u = ri.data;
+        sds line = sdsnew("user ");
+        line = sdscatsds(line,u->name);
+        line = sdscatlen(line," ",1);
+        sds descr = ACLDescribeUser(u);
+        line = sdscatsds(line,descr);
+        sdsfree(descr);
+        rewriteConfigRewriteLine(state,"user",line,1);
+    }
+    raxStop(&ri);
+
+    /* Mark "user" as processed in case there are no defined users. */
+    rewriteConfigMarkAsProcessed(state,"user");
+}
+
 /* Rewrite the dir option, always using absolute paths.*/
 void rewriteConfigDirOption(struct rewriteConfigState *state) {
     char cwd[1024];
@@ -2174,10 +2220,12 @@ int rewriteConfig(char *path) {
     rewriteConfigNumericalOption(state,"replica-announce-port",server.slave_announce_port,CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT);
     rewriteConfigEnumOption(state,"loglevel",server.verbosity,loglevel_enum,CONFIG_DEFAULT_VERBOSITY);
     rewriteConfigStringOption(state,"logfile",server.logfile,CONFIG_DEFAULT_LOGFILE);
+    rewriteConfigStringOption(state,"aclfile",server.acl_filename,CONFIG_DEFAULT_ACL_FILENAME);
     rewriteConfigYesNoOption(state,"syslog-enabled",server.syslog_enabled,CONFIG_DEFAULT_SYSLOG_ENABLED);
     rewriteConfigStringOption(state,"syslog-ident",server.syslog_ident,CONFIG_DEFAULT_SYSLOG_IDENT);
     rewriteConfigSyslogfacilityOption(state);
     rewriteConfigSaveOption(state);
+    rewriteConfigUserOption(state);
     rewriteConfigNumericalOption(state,"databases",server.dbnum,CONFIG_DEFAULT_DBNUM);
     rewriteConfigYesNoOption(state,"stop-writes-on-bgsave-error",server.stop_writes_on_bgsave_err,CONFIG_DEFAULT_STOP_WRITES_ON_BGSAVE_ERROR);
     rewriteConfigYesNoOption(state,"rdbcompression",server.rdb_compression,CONFIG_DEFAULT_RDB_COMPRESSION);

src/module.c

@@ -684,6 +684,7 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c
     cp->rediscmd->calls = 0;
     dictAdd(server.commands,sdsdup(cmdname),cp->rediscmd);
     dictAdd(server.orig_commands,sdsdup(cmdname),cp->rediscmd);
+    cp->rediscmd->id = ACLGetCommandID(cmdname); /* ID used for ACL. */
     return REDISMODULE_OK;
 }
 
@@ -2696,6 +2697,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
     /* Create the client and dispatch the command. */
     va_start(ap, fmt);
     c = createClient(-1);
+    c->puser = NULL; /* Root user. */
     argv = moduleCreateArgvFromUserFormat(cmdname,fmt,&argc,&flags,ap);
     replicate = flags & REDISMODULE_ARGV_REPLICATE;
     va_end(ap);
@@ -4659,6 +4661,7 @@ void moduleInitModulesSystem(void) {
     moduleKeyspaceSubscribers = listCreate();
     moduleFreeContextReusedClient = createClient(-1);
     moduleFreeContextReusedClient->flags |= CLIENT_MODULE;
+    moduleFreeContextReusedClient->puser = NULL; /* root user. */
 
     moduleRegisterCoreAPI();
     if (pipe(server.module_blocked_pipe) == -1) {

src/rdb-s3.cpp

@@ -2,12 +2,42 @@ extern "C" {
 #include "rio.h"
 #include "server.h"
 }
+#include <unistd.h>
+#include <sys/wait.h>
 
 /* Save the DB on disk. Return C_ERR on error, C_OK on success. */
 extern "C" int rdbSaveS3(char *s3bucket, rdbSaveInfo *rsi)
 {
-    (void)s3bucket;
+    int fd[2];
-    (void)rsi;
+    if (pipe(fd) != 0)
+        return C_ERR;
+
+    pid_t pid = fork();
+    if (pid < 0)
+    {
+        close(fd[0]);
+        close(fd[1]);
+        return C_ERR;
+    }
+
+    if (pid == 0)
+    {
+        // child process
+        dup2(fd[1], STDIN_FILENO);
+        execlp("aws", "s3", "cp", "-", s3bucket, nullptr);
+        exit(EXIT_FAILURE);
+    }
+    else
+    {
+        close(fd[1]);
+        rdbSaveFd(fd[0], rsi);
+        int status;
+        waitpid(pid, &status, 0);
+    }
+    
+    close(fd[0]);
+    
+    
     // NOP
     return C_ERR;
 }

src/rdb.h

@@ -141,6 +141,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
 void rdbRemoveTempFile(pid_t childpid);
 int rdbSave(char *filename, rdbSaveInfo *rsi);
 int rdbSaveFd(int fd, rdbSaveInfo *rsi);
+int rdbSaveS3(char *path, rdbSaveInfo *rsi);
 ssize_t rdbSaveObject(rio *rdb, robj *o);
 size_t rdbSavedObjectLen(robj *o);
 robj *rdbLoadObject(int type, rio *rdb);

src/scripting.c

@@ -460,6 +460,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     /* Setup our fake client for command execution */
     c->argv = argv;
     c->argc = argc;
+    c->puser = server.lua_caller->puser;
 
     /* Log the command if debugging is active. */
     if (ldb.active && ldb.step) {
@@ -497,6 +498,19 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
         goto cleanup;
     }
 
+    /* Check the ACLs. */
+    int acl_retval = ACLCheckCommandPerm(c);
+    if (acl_retval != ACL_OK) {
+        if (acl_retval == ACL_DENIED_CMD)
+            luaPushError(lua, "The user executing the script can't run this "
+                              "command or subcommand");
+        else
+            luaPushError(lua, "The user executing the script can't access "
+                              "at least one of the keys mentioned in the "
+                              "command arguments");
+        goto cleanup;
+    }
+
     /* Write commands are forbidden against read-only slaves, or if a
      * command marked as non-deterministic was already called in the context
      * of this script. */
@@ -655,6 +669,8 @@ cleanup:
         argv_size = 0;
     }
 
+    c->puser = NULL;
+
     if (raise_error) {
         /* If we are here we should have an error in the stack, in the
          * form of a table with an "err" field. Extract the string to

src/server.c

@@ -2269,6 +2269,7 @@ void initServerConfig(void) {
     server.rdb_filename = zstrdup(CONFIG_DEFAULT_RDB_FILENAME);
     server.rdb_s3bucketpath = NULL;
     server.aof_filename = zstrdup(CONFIG_DEFAULT_AOF_FILENAME);
+    server.acl_filename = zstrdup(CONFIG_DEFAULT_ACL_FILENAME);
     server.rdb_compression = CONFIG_DEFAULT_RDB_COMPRESSION;
     server.rdb_checksum = CONFIG_DEFAULT_RDB_CHECKSUM;
     server.stop_writes_on_bgsave_err = CONFIG_DEFAULT_STOP_WRITES_ON_BGSAVE_ERROR;
@@ -4911,6 +4912,11 @@ int main(int argc, char **argv) {
         linuxMemoryWarnings();
     #endif
         moduleLoadFromQueue();
+        if (ACLLoadConfiguredUsers() == C_ERR) {
+            serverLog(LL_WARNING,
+                "Critical error while loading ACLs. Exiting.");
+            exit(1);
+        }
         loadDataFromDisk();
         if (server.cluster_enabled) {
             if (verifyClusterConfigWithData() == C_ERR) {

src/server.h

@@ -148,6 +148,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CONFIG_DEFAULT_RDB_SAVE_INCREMENTAL_FSYNC 1
 #define CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE 0
 #define CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG 10
+#define CONFIG_DEFAULT_ACL_FILENAME ""
 #define NET_IP_STR_LEN 46 /* INET6_ADDRSTRLEN is 46, but we need to be sure */
 #define NET_PEER_ID_LEN (NET_IP_STR_LEN+32) /* Must be enough for ip:port */
 #define CONFIG_BINDADDR_MAX 16
@@ -740,9 +741,10 @@ typedef struct readyList {
                                            command ID we can set in the user
                                            is USER_COMMAND_BITS_COUNT-1. */
 #define USER_FLAG_ENABLED (1<<0)        /* The user is active. */
-#define USER_FLAG_ALLKEYS (1<<1)        /* The user can mention any key. */
+#define USER_FLAG_DISABLED (1<<1)       /* The user is disabled. */
-#define USER_FLAG_ALLCOMMANDS (1<<2)    /* The user can run all commands. */
+#define USER_FLAG_ALLKEYS (1<<2)        /* The user can mention any key. */
-#define USER_FLAG_NOPASS      (1<<3)    /* The user requires no password, any
+#define USER_FLAG_ALLCOMMANDS (1<<3)    /* The user can run all commands. */
+#define USER_FLAG_NOPASS      (1<<4)    /* The user requires no password, any
                                            provided password will work. For the
                                            default user, this also means that
                                            no AUTH is needed, and every
@@ -1337,6 +1339,8 @@ struct redisServer {
     /* Latency monitor */
     long long latency_monitor_threshold;
     dict *latency_events;
+    /* ACLs */
+    char *acl_filename;     /* ACL Users file. NULL if not configured. */
     /* Assert & bug reporting */
     const char *assert_failed;
     const char *assert_file;
@@ -1725,6 +1729,7 @@ void sendChildInfo(int process_type);
 void receiveChildInfo(void);
 
 /* acl.c -- Authentication related prototypes. */
+extern rax *Users;
 extern user *DefaultUser;
 void ACLInit(void);
 /* Return values for ACLCheckUserCredentials(). */
@@ -1738,6 +1743,10 @@ int ACLCheckCommandPerm(client *c);
 int ACLSetUser(user *u, const char *op, ssize_t oplen);
 sds ACLDefaultUserFirstPassword(void);
 uint64_t ACLGetCommandCategoryFlagByName(const char *name);
+int ACLAppendUserForLoading(sds *argv, int argc, int *argc_err);
+char *ACLSetUserStringError(void);
+int ACLLoadConfiguredUsers(void);
+sds ACLDescribeUser(user *u);
 
 /* Sorted sets data type */
 

tests/unit/acl.tcl

@@ -86,4 +86,26 @@ start_server {tags {"acl"}} {
         catch {r CLIENT KILL type master} e
         set e
     } {*NOPERM*}
+
+    # Note that the order of the generated ACL rules is not stable in Redis
+    # so we need to match the different parts and not as a whole string.
+    test {ACL GETUSER is able to translate back command permissions} {
+        # Subtractive
+        r ACL setuser newuser reset +@all ~* -@string +incr -debug +debug|digest
+        set cmdstr [dict get [r ACL getuser newuser] commands]
+        assert_match {*+@all*} $cmdstr
+        assert_match {*-@string*} $cmdstr
+        assert_match {*+incr*} $cmdstr
+        assert_match {*-debug +debug|digest**} $cmdstr
+
+        # Additive
+        r ACL setuser newuser reset +@string -incr +acl +debug|digest +debug|segfault
+        set cmdstr [dict get [r ACL getuser newuser] commands]
+        assert_match {*-@all*} $cmdstr
+        assert_match {*+@string*} $cmdstr
+        assert_match {*-incr*} $cmdstr
+        assert_match {*+debug|digest*} $cmdstr
+        assert_match {*+debug|segfault*} $cmdstr
+        assert_match {*+acl*} $cmdstr
+    }
 }