ACL V2 - Selectors and key based permissions (#9974)

* Implemented selectors which provide multiple different sets of permissions to users
* Implemented key based permissions 
* Added a new ACL dry-run command to test permissions before execution
* Updated module APIs to support checking key based permissions

Co-authored-by: Oran Agra <oran@redislabs.com>

21 files changed, 1791 insertions, 596 deletions

diff --git a/redis.conf b/redis.conf
index a433e4fc4..c31203ecc 100644
--- a/redis.conf
+++ b/redis.conf
@@ -871,6 +871,10 @@ replica-priority 100
 #               commands. For instance ~* allows all the keys. The pattern
 #               is a glob-style pattern like the one of KEYS.
 #               It is possible to specify multiple patterns.
+# %R~<pattern>  Add key read pattern that specifies which keys can be read 
+#               from.
+# %W~<pattern>  Add key write pattern that specifies which keys can be
+#               written to. 
 #  allkeys      Alias for ~*
 #  resetkeys    Flush the list of allowed keys patterns.
 #  &<pattern>   Add a glob-style pattern of Pub/Sub channels that can be
@@ -896,6 +900,14 @@ replica-priority 100
 #  reset        Performs the following actions: resetpass, resetkeys, off,
 #               -@all. The user returns to the same state it has immediately
 #               after its creation.
+# (<options>)   Create a new selector with the options specified within the
+#               parentheses and attach it to the user. Each option should be 
+#               space separated. The first character must be ( and the last 
+#               character must be ).
+# clearselectors            Remove all of the currently attached selectors. 
+#                           Note this does not change the "root" user permissions,
+#                           which are the permissions directly applied onto the
+#                           user (outside the parentheses).
 #
 # ACL rules can be specified in any order: for instance you can start with
 # passwords, then flags, or key patterns. However note that the additive
diff --git a/src/acl.c b/src/acl.c
index aff48afd4..d4fa2c6c8 100644
--- a/src/acl.c
+++ b/src/acl.c
@@ -92,19 +92,59 @@ struct ACLUserFlag {
     /* Note: the order here dictates the emitted order at ACLDescribeUser */
     {"on", USER_FLAG_ENABLED},
     {"off", USER_FLAG_DISABLED},
-    {"allkeys", USER_FLAG_ALLKEYS},
-    {"allchannels", USER_FLAG_ALLCHANNELS},
-    {"allcommands", USER_FLAG_ALLCOMMANDS},
     {"nopass", USER_FLAG_NOPASS},
     {"skip-sanitize-payload", USER_FLAG_SANITIZE_PAYLOAD_SKIP},
     {"sanitize-payload", USER_FLAG_SANITIZE_PAYLOAD},
     {NULL,0} /* Terminator. */
 };
 
-void ACLResetFirstArgsForCommand(user *u, unsigned long id);
-void ACLResetFirstArgs(user *u);
-void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub);
+struct ACLSelectorFlags {
+    const char *name;
+    uint64_t flag;
+} ACLSelectorFlags[] = {
+    /* Note: the order here dictates the emitted order at ACLDescribeUser */
+    {"allkeys", SELECTOR_FLAG_ALLKEYS},
+    {"allchannels", SELECTOR_FLAG_ALLCHANNELS},
+    {"allcommands", SELECTOR_FLAG_ALLCOMMANDS},
+    {NULL,0} /* Terminator. */
+};
+
+/* ACL selectors are private and not exposed outside of acl.c. */
+typedef struct {
+    uint32_t flags; /* See SELECTOR_FLAG_* */
+    /* The bit in allowed_commands is set if this user has the right to
+     * execute this command.
+     *
+     * If the bit for a given command is NOT set and the command has
+     * allowed first-args, Redis will also check allowed_firstargs in order to
+     * understand if the command can be executed. */
+    uint64_t allowed_commands[USER_COMMAND_BITS_COUNT/64];
+    /* allowed_firstargs is used by ACL rules to block access to a command unless a
+     * specific argv[1] is given (or argv[2] in case it is applied on a sub-command).
+     * For example, a user can use the rule "-select +select|0" to block all
+     * SELECT commands, except "SELECT 0".
+     * And for a sub-command: "+config -config|set +config|set|loglevel"
+     *
+     * For each command ID (corresponding to the command bit set in allowed_commands),
+     * This array points to an array of SDS strings, terminated by a NULL pointer,
+     * with all the first-args that are allowed for this command. When no first-arg
+     * matching is used, the field is just set to NULL to avoid allocating
+     * USER_COMMAND_BITS_COUNT pointers. */
+    sds **allowed_firstargs;
+    list *patterns;  /* A list of allowed key patterns. If this field is NULL
+                        the user cannot mention any key in a command, unless
+                        the flag ALLKEYS is set in the user. */
+    list *channels;  /* A list of allowed Pub/Sub channel patterns. If this
+                        field is NULL the user cannot mention any channel in a
+                        `PUBLISH` or [P][UNSUBSCRIBE] command, unless the flag
+                        ALLCHANNELS is set in the user. */
+} aclSelector;
+
+void ACLResetFirstArgsForCommand(aclSelector *selector, unsigned long id);
+void ACLResetFirstArgs(aclSelector *selector);
+void ACLAddAllowedFirstArg(aclSelector *selector, unsigned long id, const char *sub);
 void ACLFreeLogEntry(void *le);
+int ACLSetSelector(aclSelector *selector, const char *op, size_t oplen);
 
 /* The length of the string representation of a hashed password. */
 #define HASH_PASSWORD_LEN SHA256_BLOCK_SIZE*2
@@ -244,6 +284,128 @@ void *ACLListDupSds(void *item) {
     return sdsdup(item);
 }
 
+/* Structure used for handling key patterns with different key
+ * based permissions. */
+typedef struct {
+    int flags; /* The CMD_KEYS_* flags for this key pattern */
+    sds pattern; /* The pattern to match keys against */
+} keyPattern;
+
+/* Create a new key pattern. */
+keyPattern *ACLKeyPatternCreate(sds pattern, int flags) {
+    keyPattern *new = (keyPattern *) zmalloc(sizeof(keyPattern));
+    new->pattern = pattern;
+    new->flags = flags;
+    return new;
+}
+
+/* Free a key pattern and internal structures. */
+void ACLKeyPatternFree(keyPattern *pattern) {
+    sdsfree(pattern->pattern);
+    zfree(pattern);
+}
+
+/* Method for passwords/pattern comparison used for the user->passwords list
+ * so that we can search for items with listSearchKey(). */
+int ACLListMatchKeyPattern(void *a, void *b) {
+    return sdscmp(((keyPattern *) a)->pattern,((keyPattern *) b)->pattern) == 0;
+}
+
+/* Method to free list elements from ACL users password/patterns lists. */
+void ACLListFreeKeyPattern(void *item) {
+    ACLKeyPatternFree(item);
+}
+
+/* Method to duplicate list elements from ACL users password/patterns lists. */
+void *ACLListDupKeyPattern(void *item) {
+    keyPattern *old = (keyPattern *) item;
+    return ACLKeyPatternCreate(sdsdup(old->pattern), old->flags);
+}
+
+/* Append the string representation of a key pattern onto the
+ * provided base string. */
+sds sdsCatPatternString(sds base, keyPattern *pat) {
+    if (pat->flags == ACL_ALL_PERMISSION) {
+        base = sdscatlen(base,"~",1);
+    } else if (pat->flags == ACL_READ_PERMISSION) {
+        base = sdscatlen(base,"%R~",3);
+    } else if (pat->flags == ACL_WRITE_PERMISSION) {
+        base = sdscatlen(base,"%W~",3);
+    } else {
+        serverPanic("Invalid key pattern flag detected");
+    }
+    return sdscatsds(base, pat->pattern);
+}
+
+/* Create an empty selector with the provided set of initial
+ * flags. The selector will be default have no permissions. */
+aclSelector *ACLCreateSelector(int flags) {
+    aclSelector *selector = zmalloc(sizeof(aclSelector));
+    selector->flags = flags | server.acl_pubsub_default;
+    selector->patterns = listCreate();
+    selector->channels = listCreate();
+    selector->allowed_firstargs = NULL;
+
+    listSetMatchMethod(selector->patterns,ACLListMatchKeyPattern);
+    listSetFreeMethod(selector->patterns,ACLListFreeKeyPattern);
+    listSetDupMethod(selector->patterns,ACLListDupKeyPattern);
+    listSetMatchMethod(selector->channels,ACLListMatchSds);
+    listSetFreeMethod(selector->channels,ACLListFreeSds);
+    listSetDupMethod(selector->channels,ACLListDupSds);
+    memset(selector->allowed_commands,0,sizeof(selector->allowed_commands));
+
+    return selector;
+}
+
+/* Cleanup the provided selector, including all interior structures. */
+void ACLFreeSelector(aclSelector *selector) {
+    listRelease(selector->patterns);
+    listRelease(selector->channels);
+    ACLResetFirstArgs(selector);
+    zfree(selector);
+}
+
+/* Create an exact copy of the provided selector. */
+aclSelector *ACLCopySelector(aclSelector *src) {
+    aclSelector *dst = zmalloc(sizeof(aclSelector));
+    dst->flags = src->flags;
+    dst->patterns = listDup(src->patterns);
+    dst->channels = listDup(src->channels);
+    memcpy(dst->allowed_commands,src->allowed_commands,
+           sizeof(dst->allowed_commands));
+    dst->allowed_firstargs = NULL;
+    /* Copy the allowed first-args array of array of SDS strings. */
+    if (src->allowed_firstargs) {
+        for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
+            if (!(src->allowed_firstargs[j])) continue;
+            for (int i = 0; src->allowed_firstargs[j][i]; i++) {
+                ACLAddAllowedFirstArg(dst, j, src->allowed_firstargs[j][i]);
+            }
+        }
+    }
+    return dst;
+}
+
+/* List method for freeing a selector */
+void ACLListFreeSelector(void *a) {
+    ACLFreeSelector((aclSelector *) a);
+}
+
+/* List method for duplicating a selector */
+void *ACLListDuplicateSelector(void *src) {
+    return ACLCopySelector((aclSelector *)src);
+}
+
+/* All users have an implicit root selector which
+ * provides backwards compatibility to the old ACLs-
+ * permissions. */
+aclSelector *ACLUserGetRootSelector(user *u) {
+    serverAssert(listLength(u->selectors));
+    aclSelector *s = (aclSelector *) listNodeValue(listFirst(u->selectors));
+    serverAssert(s->flags & SELECTOR_FLAG_ROOT);
+    return s;
+}
+
 /* Create a new user with the specified name, store it in the list
  * of users (the Users global radix tree), and returns a reference to
  * the structure representing the user.
@@ -253,21 +415,20 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     if (raxFind(Users,(unsigned char*)name,namelen) != raxNotFound) return NULL;
     user *u = zmalloc(sizeof(*u));
     u->name = sdsnewlen(name,namelen);
-    u->flags = USER_FLAG_DISABLED | server.acl_pubsub_default;
-    u->allowed_firstargs = NULL;
+    u->flags = USER_FLAG_DISABLED;
     u->passwords = listCreate();
-    u->patterns = listCreate();
-    u->channels = listCreate();
     listSetMatchMethod(u->passwords,ACLListMatchSds);
     listSetFreeMethod(u->passwords,ACLListFreeSds);
     listSetDupMethod(u->passwords,ACLListDupSds);
-    listSetMatchMethod(u->patterns,ACLListMatchSds);
-    listSetFreeMethod(u->patterns,ACLListFreeSds);
-    listSetDupMethod(u->patterns,ACLListDupSds);
-    listSetMatchMethod(u->channels,ACLListMatchSds);
-    listSetFreeMethod(u->channels,ACLListFreeSds);
-    listSetDupMethod(u->channels,ACLListDupSds);
-    memset(u->allowed_commands,0,sizeof(u->allowed_commands));
+
+    u->selectors = listCreate();
+    listSetFreeMethod(u->selectors,ACLListFreeSelector);
+    listSetDupMethod(u->selectors,ACLListDuplicateSelector);
+
+    /* Add the initial root selector */
+    aclSelector *s = ACLCreateSelector(SELECTOR_FLAG_ROOT);
+    listAddNodeHead(u->selectors, s);
+
     raxInsert(Users,(unsigned char*)name,namelen,u,NULL);
     return u;
 }
@@ -294,9 +455,7 @@ user *ACLCreateUnlinkedUser(void) {
 void ACLFreeUser(user *u) {
     sdsfree(u->name);
     listRelease(u->passwords);
-    listRelease(u->patterns);
-    listRelease(u->channels);
-    ACLResetFirstArgs(u);
+    listRelease(u->selectors);
     zfree(u);
 }
 
@@ -335,27 +494,10 @@ void ACLFreeUserAndKillClients(user *u) {
  * same rules (but the names will continue to be the original ones). */
 void ACLCopyUser(user *dst, user *src) {
     listRelease(dst->passwords);
-    listRelease(dst->patterns);
-    listRelease(dst->channels);
+    listRelease(dst->selectors);
     dst->passwords = listDup(src->passwords);
-    dst->patterns = listDup(src->patterns);
-    dst->channels = listDup(src->channels);
-    memcpy(dst->allowed_commands,src->allowed_commands,
-           sizeof(dst->allowed_commands));
+    dst->selectors = listDup(src->selectors);
     dst->flags = src->flags;
-    ACLResetFirstArgs(dst);
-    /* Copy the allowed first-args array of array of SDS strings. */
-    if (src->allowed_firstargs) {
-        for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
-            if (src->allowed_firstargs[j]) {
-                for (int i = 0; src->allowed_firstargs[j][i]; i++)
-                {
-                    ACLAddAllowedFirstArg(dst, j,
-                        src->allowed_firstargs[j][i]);
-                }
-            }
-        }
-    }
 }
 
 /* Free all the users registered in the radix tree 'users' and free the
@@ -384,17 +526,17 @@ int ACLGetCommandBitCoordinates(uint64_t id, uint64_t *word, uint64_t *bit) {
  *
  * If the bit overflows the user internal representation, zero is returned
  * in order to disallow the execution of the command in such edge case. */
-int ACLGetUserCommandBit(const user *u, unsigned long id) {
+int ACLGetSelectorCommandBit(const aclSelector *selector, unsigned long id) {
     uint64_t word, bit;
     if (ACLGetCommandBitCoordinates(id,&word,&bit) == C_ERR) return 0;
-    return (u->allowed_commands[word] & bit) != 0;
+    return (selector->allowed_commands[word] & bit) != 0;
 }
 
 /* When +@all or allcommands is given, we set a reserved bit as well that we
  * can later test, to see if the user has the right to execute "future commands",
  * that is, commands loaded later via modules. */
-int ACLUserCanExecuteFutureCommands(user *u) {
-    return ACLGetUserCommandBit(u,USER_COMMAND_BITS_COUNT-1);
+int ACLSelectorCanExecuteFutureCommands(aclSelector *selector) {
+    return ACLGetSelectorCommandBit(selector,USER_COMMAND_BITS_COUNT-1);
 }
 
 /* Set the specified command bit for the specified user to 'value' (0 or 1).
@@ -402,76 +544,76 @@ int ACLUserCanExecuteFutureCommands(user *u) {
  * is performed. As a side effect of calling this function with a value of
  * zero, the user flag ALLCOMMANDS is cleared since it is no longer possible
  * to skip the command bit explicit test. */
-void ACLSetUserCommandBit(user *u, unsigned long id, int value) {
+void ACLSetSelectorCommandBit(aclSelector *selector, unsigned long id, int value) {
     uint64_t word, bit;
     if (ACLGetCommandBitCoordinates(id,&word,&bit) == C_ERR) return;
     if (value) {
-        u->allowed_commands[word] |= bit;
+        selector->allowed_commands[word] |= bit;
     } else {
-        u->allowed_commands[word] &= ~bit;
-        u->flags &= ~USER_FLAG_ALLCOMMANDS;
+        selector->allowed_commands[word] &= ~bit;
+        selector->flags &= ~SELECTOR_FLAG_ALLCOMMANDS;
     }
 }
 
 /* This function is used to allow/block a specific command.
  * Allowing/blocking a container command also applies for its subcommands */
-void ACLChangeCommandPerm(user *u, struct redisCommand *cmd, int allow) {
+void ACLChangeSelectorPerm(aclSelector *selector, struct redisCommand *cmd, int allow) {
     unsigned long id = cmd->id;
-    ACLSetUserCommandBit(u,id,allow);
-    ACLResetFirstArgsForCommand(u,id);
+    ACLSetSelectorCommandBit(selector,id,allow);
+    ACLResetFirstArgsForCommand(selector,id);
     if (cmd->subcommands_dict) {
         dictEntry *de;
         dictIterator *di = dictGetSafeIterator(cmd->subcommands_dict);
         while((de = dictNext(di)) != NULL) {
             struct redisCommand *sub = (struct redisCommand *)dictGetVal(de);
-            ACLSetUserCommandBit(u,sub->id,allow);
+            ACLSetSelectorCommandBit(selector,sub->id,allow);
         }
         dictReleaseIterator(di);
     }
 }
 
-void ACLSetUserCommandBitsForCategoryLogic(dict *commands, user *u, uint64_t cflag, int value) {
+void ACLSetSelectorCommandBitsForCategoryLogic(dict *commands, aclSelector *selector, uint64_t cflag, int value) {
     dictIterator *di = dictGetIterator(commands);
     dictEntry *de;
     while ((de = dictNext(di)) != NULL) {
         struct redisCommand *cmd = dictGetVal(de);
         if (cmd->flags & CMD_MODULE) continue; /* Ignore modules commands. */
         if (cmd->acl_categories & cflag) {
-            ACLChangeCommandPerm(u,cmd,value);
+            ACLChangeSelectorPerm(selector,cmd,value);
         }
         if (cmd->subcommands_dict) {
-            ACLSetUserCommandBitsForCategoryLogic(cmd->subcommands_dict, u, cflag, value);
+            ACLSetSelectorCommandBitsForCategoryLogic(cmd->subcommands_dict, selector, cflag, value);
         }
     }
     dictReleaseIterator(di);
 }
 
-/* This is like ACLSetUserCommandBit(), but instead of setting the specified
+/* This is like ACLSetSelectorCommandBit(), but instead of setting the specified
  * ID, it will check all the commands in the category specified as argument,
  * and will set all the bits corresponding to such commands to the specified
  * value. Since the category passed by the user may be non existing, the
  * function returns C_ERR if the category was not found, or C_OK if it was
  * found and the operation was performed. */
-int ACLSetUserCommandBitsForCategory(user *u, const char *category, int value) {
+int ACLSetSelectorCommandBitsForCategory(aclSelector *selector, const char *category, int value) {
     uint64_t cflag = ACLGetCommandCategoryFlagByName(category);
     if (!cflag) return C_ERR;
-    ACLSetUserCommandBitsForCategoryLogic(server.orig_commands, u, cflag, value);
+    ACLSetSelectorCommandBitsForCategoryLogic(server.orig_commands, selector, cflag, value);
     return C_OK;
 }
 
-void ACLCountCategoryBitsForCommands(dict *commands, user *u, unsigned long *on, unsigned long *off, uint64_t cflag) {
+void ACLCountCategoryBitsForCommands(dict *commands, aclSelector *selector, unsigned long *on, unsigned long *off, uint64_t cflag) {
     dictIterator *di = dictGetIterator(commands);
     dictEntry *de;
     while ((de = dictNext(di)) != NULL) {
         struct redisCommand *cmd = dictGetVal(de);
         if (cmd->acl_categories & cflag) {
-            if (ACLGetUserCommandBit(u,cmd->id))
+            if (ACLGetSelectorCommandBit(selector,cmd->id))
                 (*on)++;
             else
                 (*off)++;
         }
         if (cmd->subcommands_dict) {
-            ACLCountCategoryBitsForCommands(cmd->subcommands_dict, u, on, off, cflag);
+            ACLCountCategoryBitsForCommands(cmd->subcommands_dict, selector, on, off, cflag);
         }
     }
     dictReleaseIterator(di);
@@ -481,47 +623,48 @@ void ACLCountCategoryBitsForCommands(dict *commands, user *u, unsigned long *on,
  * in the subset of commands flagged with the specified category name.
  * If the category name is not valid, C_ERR is returned, otherwise C_OK is
  * returned and on and off are populated by reference. */
-int ACLCountCategoryBitsForUser(user *u, unsigned long *on, unsigned long *off,
+int ACLCountCategoryBitsForSelector(aclSelector *selector, unsigned long *on, unsigned long *off,
                                 const char *category)
 {
     uint64_t cflag = ACLGetCommandCategoryFlagByName(category);
     if (!cflag) return C_ERR;
 
     *on = *off = 0;
-    ACLCountCategoryBitsForCommands(server.orig_commands, u, on, off, cflag);
+    ACLCountCategoryBitsForCommands(server.orig_commands, selector, on, off, cflag);
     return C_OK;
 }
 
-sds ACLDescribeUserCommandRulesSingleCommands(user *u, user *fakeuser, sds rules, dict *commands) {
+sds ACLDescribeSelectorCommandRulesSingleCommands(aclSelector *selector, aclSelector *fake_selector,
+        sds rules, dict *commands) {
     dictIterator *di = dictGetIterator(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);
+        int userbit = ACLGetSelectorCommandBit(selector,cmd->id);
+        int fakebit = ACLGetSelectorCommandBit(fake_selector,cmd->id);
         if (userbit != fakebit) {
             rules = sdscatlen(rules, userbit ? "+" : "-", 1);
             sds fullname = getFullCommandName(cmd);
             rules = sdscat(rules,fullname);
             sdsfree(fullname);
             rules = sdscatlen(rules," ",1);
-            ACLChangeCommandPerm(fakeuser,cmd,userbit);
+            ACLChangeSelectorPerm(fake_selector,cmd,userbit);
         }
 
         if (cmd->subcommands_dict)
-            rules = ACLDescribeUserCommandRulesSingleCommands(u,fakeuser,rules,cmd->subcommands_dict);
+            rules = ACLDescribeSelectorCommandRulesSingleCommands(selector,fake_selector,rules,cmd->subcommands_dict);
 
         /* Emit the first-args if there are any. */
-        if (userbit == 0 && u->allowed_firstargs &&
-            u->allowed_firstargs[cmd->id])
+        if (userbit == 0 && selector->allowed_firstargs &&
+            selector->allowed_firstargs[cmd->id])
         {
-            for (int j = 0; u->allowed_firstargs[cmd->id][j]; j++) {
+            for (int j = 0; selector->allowed_firstargs[cmd->id][j]; j++) {
                 rules = sdscatlen(rules,"+",1);
                 sds fullname = getFullCommandName(cmd);
                 rules = sdscat(rules,fullname);
                 sdsfree(fullname);
                 rules = sdscatlen(rules,"|",1);
-                rules = sdscatsds(rules,u->allowed_firstargs[cmd->id][j]);
+                rules = sdscatsds(rules,selector->allowed_firstargs[cmd->id][j]);
                 rules = sdscatlen(rules," ",1);
             }
         }
@@ -530,14 +673,14 @@ sds ACLDescribeUserCommandRulesSingleCommands(user *u, user *fakeuser, sds rules
     return rules;
 }
 
-/* This function returns an SDS string representing the specified user ACL
+/* This function returns an SDS string representing the specified selector 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 ACLDescribeSelectorCommandRules(aclSelector *selector) {
     sds rules = sdsempty();
     int additive;   /* If true we start from -@all and add, otherwise if
                        false we start from +@all and remove. */
@@ -545,8 +688,8 @@ sds ACLDescribeUserCommandRules(user *u) {
     /* 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;
+    aclSelector fs = {0};
+    aclSelector *fake_selector = &fs;
 
     /* 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
@@ -556,14 +699,14 @@ sds ACLDescribeUserCommandRules(user *u) {
      * 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)) {
+    if (ACLSelectorCanExecuteFutureCommands(selector)) {
         additive = 0;
         rules = sdscat(rules,"+@all ");
-        ACLSetUser(fakeuser,"+@all",-1);
+        ACLSetSelector(fake_selector,"+@all",-1);
     } else {
         additive = 1;
         rules = sdscat(rules,"-@all ");
-        ACLSetUser(fakeuser,"-@all",-1);
+        ACLSetSelector(fake_selector,"-@all",-1);
     }
 
     /* Attempt to find a good approximation for categories and commands
@@ -573,17 +716,17 @@ sds ACLDescribeUserCommandRules(user *u) {
      * final pass adding/removing the single commands needed to make the bitmap
      * exactly match. A temp user is maintained to keep track of categories 
      * already applied. */
-    user tu = {0};
-    user *tempuser = &tu;
+    aclSelector ts = {0};
+    aclSelector *temp_selector = &ts;
     
     /* Keep track of the categories that have been applied, to prevent
      * applying them twice.  */
     char applied[sizeof(ACLCommandCategories)/sizeof(ACLCommandCategories[0])];
     memset(applied, 0, sizeof(applied));
 
-    memcpy(tempuser->allowed_commands,
-        u->allowed_commands, 
-        sizeof(u->allowed_commands));
+    memcpy(temp_selector->allowed_commands,
+        selector->allowed_commands, 
+        sizeof(selector->allowed_commands));
     while (1) {
         int best = -1;
         unsigned long mindiff = INT_MAX, maxsame = 0;
@@ -591,7 +734,7 @@ sds ACLDescribeUserCommandRules(user *u) {
             if (applied[j]) continue;
 
             unsigned long on, off, diff, same;
-            ACLCountCategoryBitsForUser(tempuser,&on,&off,ACLCommandCategories[j].name);
+            ACLCountCategoryBitsForSelector(temp_selector,&on,&off,ACLCommandCategories[j].name);
             /* Check if the current category is the best this loop:
              * * It has more commands in common with the user than commands
              *   that are different.
@@ -616,11 +759,11 @@ sds ACLDescribeUserCommandRules(user *u) {
 
         sds op = sdsnewlen(additive ? "+@" : "-@", 2);
         op = sdscat(op,ACLCommandCategories[best].name);
-        ACLSetUser(fakeuser,op,-1);
+        ACLSetSelector(fake_selector,op,-1);
 
         sds invop = sdsnewlen(additive ? "-@" : "+@", 2);
         invop = sdscat(invop,ACLCommandCategories[best].name);
-        ACLSetUser(tempuser,invop,-1);
+        ACLSetSelector(temp_selector,invop,-1);
 
         rules = sdscatsds(rules,op);
         rules = sdscatlen(rules," ",1);
@@ -631,7 +774,7 @@ sds ACLDescribeUserCommandRules(user *u) {
     }
 
     /* Fix the final ACLs with single commands differences. */
-    rules = ACLDescribeUserCommandRulesSingleCommands(u,fakeuser,rules,server.orig_commands);
+    rules = ACLDescribeSelectorCommandRulesSingleCommands(selector,fake_selector,rules,server.orig_commands);
 
     /* Trim the final useless space. */
     sdsrange(rules,0,-2);
@@ -640,22 +783,59 @@ sds ACLDescribeUserCommandRules(user *u) {
      * 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)
+    if (memcmp(fake_selector->allowed_commands,
+                        selector->allowed_commands,
+                        sizeof(selector->allowed_commands)) != 0)
     {
         serverLog(LL_WARNING,
             "CRITICAL ERROR: User ACLs don't match final bitmap: '%s'",
             rules);
-        serverPanic("No bitmap match in ACLDescribeUserCommandRules()");
+        serverPanic("No bitmap match in ACLDescribeSelectorCommandRules()");
     }
     return rules;
 }
 
-/* This is similar to ACLDescribeUserCommandRules(), however instead of
+sds ACLDescribeSelector(aclSelector *selector) {
+    listIter li;
+    listNode *ln;
+    sds res = sdsempty();
+    /* Key patterns. */
+    if (selector->flags & SELECTOR_FLAG_ALLKEYS) {
+        res = sdscatlen(res,"~* ",3);
+    } else {
+        listRewind(selector->patterns,&li);
+        while((ln = listNext(&li))) {
+            keyPattern *thispat = (keyPattern *)listNodeValue(ln);
+            res = sdsCatPatternString(res, thispat);
+            res = sdscatlen(res," ",1);
+        }
+    }
+
+    /* Pub/sub channel patterns. */
+    if (selector->flags & SELECTOR_FLAG_ALLCHANNELS) {
+        res = sdscatlen(res,"&* ",3);
+    } else {
+        res = sdscatlen(res,"resetchannels ",14);
+        listRewind(selector->channels,&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 = ACLDescribeSelectorCommandRules(selector);
+    res = sdscatsds(res,rules);
+    sdsfree(rules);
+    return res;
+}
+
+/* This is similar to ACLDescribeSelectorCommandRules(), 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
+ * the ACLDescribeSelectorCommandRules() 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) {
@@ -663,11 +843,6 @@ sds ACLDescribeUser(user *u) {
 
     /* Flags. */
     for (int j = 0; ACLUserFlags[j].flag; j++) {
-        /* Skip the allcommands, allkeys and allchannels flags because they'll
-         * be emitted later as +@all, ~* and &*. */
-        if (ACLUserFlags[j].flag == USER_FLAG_ALLKEYS ||
-            ACLUserFlags[j].flag == USER_FLAG_ALLCHANNELS ||
-            ACLUserFlags[j].flag == USER_FLAG_ALLCOMMANDS) continue;
         if (u->flags & ACLUserFlags[j].flag) {
             res = sdscat(res,ACLUserFlags[j].name);
             res = sdscatlen(res," ",1);
@@ -685,37 +860,18 @@ sds ACLDescribeUser(user *u) {
         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);
-        }
-    }
-
-    /* Pub/sub channel patterns. */
-    if (u->flags & USER_FLAG_ALLCHANNELS) {
-        res = sdscatlen(res,"&* ",3);
-    } else {
-        res = sdscatlen(res,"resetchannels ",14);
-        listRewind(u->channels,&li);
-        while((ln = listNext(&li))) {
-            sds thispat = listNodeValue(ln);
-            res = sdscatlen(res,"&",1);
-            res = sdscatsds(res,thispat);
-            res = sdscatlen(res," ",1);
+    /* Selectors (Commands and keys) */
+    listRewind(u->selectors,&li);
+    while((ln = listNext(&li))) {
+        aclSelector *selector = (aclSelector *) listNodeValue(ln);
+        sds default_perm = ACLDescribeSelector(selector);
+        if (selector->flags & SELECTOR_FLAG_ROOT) {
+            res = sdscatfmt(res, "%s", default_perm);
+        } else {
+            res = sdscatfmt(res, " (%s)", default_perm);
         }
+        sdsfree(default_perm);
     }
-
-    /* Command rules. */
-    sds rules = ACLDescribeUserCommandRules(u);
-    res = sdscatsds(res,rules);
-    sdsfree(rules);
     return res;
 }
 
@@ -732,38 +888,38 @@ struct redisCommand *ACLLookupCommand(const char *name) {
 
 /* Flush the array of allowed first-args for the specified user
  * and command ID. */
-void ACLResetFirstArgsForCommand(user *u, unsigned long id) {
-    if (u->allowed_firstargs && u->allowed_firstargs[id]) {
-        for (int i = 0; u->allowed_firstargs[id][i]; i++)
-            sdsfree(u->allowed_firstargs[id][i]);
-        zfree(u->allowed_firstargs[id]);
-        u->allowed_firstargs[id] = NULL;
+void ACLResetFirstArgsForCommand(aclSelector *selector, unsigned long id) {
+    if (selector->allowed_firstargs && selector->allowed_firstargs[id]) {
+        for (int i = 0; selector->allowed_firstargs[id][i]; i++)
+            sdsfree(selector->allowed_firstargs[id][i]);
+        zfree(selector->allowed_firstargs[id]);
+        selector->allowed_firstargs[id] = NULL;
     }
 }
 
 /* Flush the entire table of first-args. This is useful on +@all, -@all
  * or similar to return back to the minimal memory usage (and checks to do)
  * for the user. */
-void ACLResetFirstArgs(user *u) {
-    if (u->allowed_firstargs == NULL) return;
+void ACLResetFirstArgs(aclSelector *selector) {
+    if (selector->allowed_firstargs == NULL) return;
     for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
-        if (u->allowed_firstargs[j]) {
-            for (int i = 0; u->allowed_firstargs[j][i]; i++)
-                sdsfree(u->allowed_firstargs[j][i]);
-            zfree(u->allowed_firstargs[j]);
+        if (selector->allowed_firstargs[j]) {
+            for (int i = 0; selector->allowed_firstargs[j][i]; i++)
+                sdsfree(selector->allowed_firstargs[j][i]);
+            zfree(selector->allowed_firstargs[j]);
         }
     }
-    zfree(u->allowed_firstargs);
-    u->allowed_firstargs = NULL;
+    zfree(selector->allowed_firstargs);
+    selector->allowed_firstargs = NULL;
 }
 
 /* Add a first-arh to the list of subcommands for the user 'u' and
  * the command id specified. */
-void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
+void ACLAddAllowedFirstArg(aclSelector *selector, unsigned long id, const char *sub) {
     /* If this is the first first-arg to be configured for
      * this user, we have to allocate the first-args array. */
-    if (u->allowed_firstargs == NULL) {
-        u->allowed_firstargs = zcalloc(USER_COMMAND_BITS_COUNT * sizeof(sds*));
+    if (selector->allowed_firstargs == NULL) {
+        selector->allowed_firstargs = zcalloc(USER_COMMAND_BITS_COUNT * sizeof(sds*));
     }
 
     /* We also need to enlarge the allocation pointing to the
@@ -771,10 +927,10 @@ void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
      * To start check the current size, and while we are here
      * make sure the first-arg is not already specified inside. */
     long items = 0;
-    if (u->allowed_firstargs[id]) {
-        while(u->allowed_firstargs[id][items]) {
+    if (selector->allowed_firstargs[id]) {
+        while(selector->allowed_firstargs[id][items]) {
             /* If it's already here do not add it again. */
-            if (!strcasecmp(u->allowed_firstargs[id][items],sub))
+            if (!strcasecmp(selector->allowed_firstargs[id][items],sub))
                 return;
             items++;
         }
@@ -782,18 +938,40 @@ void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
 
     /* Now we can make space for the new item (and the null term). */
     items += 2;
-    u->allowed_firstargs[id] = zrealloc(u->allowed_firstargs[id], sizeof(sds)*items);
-    u->allowed_firstargs[id][items-2] = sdsnew(sub);
-    u->allowed_firstargs[id][items-1] = NULL;
+    selector->allowed_firstargs[id] = zrealloc(selector->allowed_firstargs[id], sizeof(sds)*items);
+    selector->allowed_firstargs[id][items-2] = sdsnew(sub);
+    selector->allowed_firstargs[id][items-1] = NULL;
 }
 
-/* Set user properties according to the string "op". The following
- * is a description of what different strings will do:
+/* Create an ACL selector from the given ACL operations, which should be 
+ * a list of space separate ACL operations that starts and ends 
+ * with parentheses.
+ *
+ * If any of the operations are invalid, NULL will be returned instead
+ * and errno will be set corresponding to the interior error. */
+aclSelector *aclCreateSelectorFromOpSet(const char *opset, size_t opsetlen) {
+    serverAssert(opset[0] == '(' && opset[opsetlen - 1] == ')');
+    aclSelector *s = ACLCreateSelector(0);
+
+    int argc = 0;
+    sds trimmed = sdsnewlen(opset + 1, opsetlen - 2);
+    sds *argv = sdssplitargs(trimmed, &argc);
+    for (int i = 0; i < argc; i++) {
+        if (ACLSetSelector(s, argv[i], sdslen(argv[i])) == C_ERR) {
+            ACLFreeSelector(s);
+            s = NULL;
+            goto cleanup;
+        }
+    }
+
+cleanup:
+    sdsfreesplitres(argv, argc);
+    sdsfree(trimmed);
+    return s;
+}
+
+/* Set a selector's properties with the provided 'op'.
  *
- * on           Enable the user: it is possible to authenticate as this user.
- * off          Disable the user: it's no longer possible to authenticate
- *              with this user, however the already authenticated connections
- *              will still work.
  * +<command>   Allow the execution of that command.
  *              May be used with `|` for allowing subcommands (e.g "+config|get")
  * -<command>   Disallow the execution of that command.
@@ -816,6 +994,10 @@ void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
  *              commands. For instance ~* allows all the keys. The pattern
  *              is a glob-style pattern like the one of KEYS.
  *              It is possible to specify multiple patterns.
+ * %R~<pattern> Add key read pattern that specifies which keys can be read 
+ *              from.
+ * %W~<pattern> Add key write pattern that specifies which keys can be
+ *              written to. 
  * allkeys      Alias for ~*
  * resetkeys    Flush the list of allowed keys patterns.
  * &<pattern>   Add a pattern of channels that can be mentioned as part of
@@ -824,6 +1006,177 @@ void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
  *              It is possible to specify multiple patterns.
  * allchannels              Alias for &*
  * resetchannels            Flush the list of allowed keys patterns.
+ */
+int ACLSetSelector(aclSelector *selector, const char* op, size_t oplen) {
+    if (!strcasecmp(op,"allkeys") ||
+               !strcasecmp(op,"~*"))
+    {
+        selector->flags |= SELECTOR_FLAG_ALLKEYS;
+        listEmpty(selector->patterns);
+    } else if (!strcasecmp(op,"resetkeys")) {
+        selector->flags &= ~SELECTOR_FLAG_ALLKEYS;
+        listEmpty(selector->patterns);
+    } else if (!strcasecmp(op,"allchannels") ||
+               !strcasecmp(op,"&*"))
+    {
+        selector->flags |= SELECTOR_FLAG_ALLCHANNELS;
+        listEmpty(selector->channels);
+    } else if (!strcasecmp(op,"resetchannels")) {
+        selector->flags &= ~SELECTOR_FLAG_ALLCHANNELS;
+        listEmpty(selector->channels);
+    } else if (!strcasecmp(op,"allcommands") ||
+               !strcasecmp(op,"+@all"))
+    {
+        memset(selector->allowed_commands,255,sizeof(selector->allowed_commands));
+        selector->flags |= SELECTOR_FLAG_ALLCOMMANDS;
+        ACLResetFirstArgs(selector);
+    } else if (!strcasecmp(op,"nocommands") ||
+               !strcasecmp(op,"-@all"))
+    {
+        memset(selector->allowed_commands,0,sizeof(selector->allowed_commands));
+        selector->flags &= ~SELECTOR_FLAG_ALLCOMMANDS;
+        ACLResetFirstArgs(selector);
+    } else if (op[0] == '~' || op[0] == '%') {
+        if (selector->flags & SELECTOR_FLAG_ALLKEYS) {
+            errno = EEXIST;
+            return C_ERR;
+        }
+        int flags = 0;
+        size_t offset = 1;
+        if (op[0] == '%') {
+            for (; offset < oplen; offset++) {
+                if (toupper(op[offset]) == 'R' && !(flags & ACL_READ_PERMISSION)) {
+                    flags |= ACL_READ_PERMISSION;
+                } else if (toupper(op[offset]) == 'W' && !(flags & ACL_WRITE_PERMISSION)) {
+                    flags |= ACL_WRITE_PERMISSION;
+                } else if (op[offset] == '~') {
+                    offset++;
+                    break;
+                } else {
+                    errno = EINVAL;
+                    return C_ERR;
+                }
+            }
+        } else {
+            flags = ACL_ALL_PERMISSION;
+        }
+
+        if (ACLStringHasSpaces(op+offset,oplen-offset)) {
+            errno = EINVAL;
+            return C_ERR;
+        }
+        keyPattern *newpat = ACLKeyPatternCreate(sdsnewlen(op+offset,oplen-offset), flags);
+        listNode *ln = listSearchKey(selector->patterns,newpat);
+        /* Avoid re-adding the same key pattern multiple times. */
+        if (ln == NULL) {
+            listAddNodeTail(selector->patterns,newpat);
+        } else {
+            ((keyPattern *)listNodeValue(ln))->flags |= flags;
+            ACLKeyPatternFree(newpat);
+        }
+        selector->flags &= ~SELECTOR_FLAG_ALLKEYS;
+    } else if (op[0] == '&') {
+        if (selector->flags & SELECTOR_FLAG_ALLCHANNELS) {
+            errno = EISDIR;
+            return C_ERR;
+        }
+        if (ACLStringHasSpaces(op+1,oplen-1)) {
+            errno = EINVAL;
+            return C_ERR;
+        }
+        sds newpat = sdsnewlen(op+1,oplen-1);
+        listNode *ln = listSearchKey(selector->channels,newpat);
+        /* Avoid re-adding the same channel pattern multiple times. */
+        if (ln == NULL)
+            listAddNodeTail(selector->channels,newpat);
+        else
+            sdsfree(newpat);
+        selector->flags &= ~SELECTOR_FLAG_ALLCHANNELS;
+    } else if (op[0] == '+' && op[1] != '@') {
+        if (strrchr(op,'|') == NULL) {
+            struct redisCommand *cmd = ACLLookupCommand(op+1);
+            if (cmd == NULL) {
+                errno = ENOENT;
+                return C_ERR;
+            }
+            ACLChangeSelectorPerm(selector,cmd,1);
+        } else {
+            /* Split the command and subcommand parts. */
+            char *copy = zstrdup(op+1);
+            char *sub = strrchr(copy,'|');
+            sub[0] = '\0';
+            sub++;
+
+            struct redisCommand *cmd = ACLLookupCommand(copy);
+
+            /* Check if the command exists. We can't check the
+             * subcommand to see if it is valid. */
+            if (cmd == NULL) {
+                zfree(copy);
+                errno = ENOENT;
+                return C_ERR;
+            }
+
+            /* The subcommand cannot be empty, so things like DEBUG|
+             * are syntax errors of course. */
+            if (strlen(sub) == 0) {
+                zfree(copy);
+                errno = EINVAL;
+                return C_ERR;
+            }
+
+            if (cmd->subcommands_dict) {
+                /* If user is trying to allow a valid subcommand we can just add its unique ID */
+                struct redisCommand *cmd = ACLLookupCommand(op+1);
+                if (cmd == NULL) {
+                    zfree(copy);
+                    errno = ENOENT;
+                    return C_ERR;
+                }
+                ACLChangeSelectorPerm(selector,cmd,1);
+            } else {
+                /* If user is trying to use the ACL mech to block SELECT except SELECT 0 or
+                 * block DEBUG except DEBUG OBJECT (DEBUG subcommands are not considered
+                 * subcommands for now) we use the allowed_firstargs mechanism. */
+                struct redisCommand *cmd = ACLLookupCommand(copy);
+                if (cmd == NULL) {
+                    zfree(copy);
+                    errno = ENOENT;
+                    return C_ERR;
+                }
+                /* Add the first-arg to the list of valid ones. */
+                ACLAddAllowedFirstArg(selector,cmd->id,sub);
+            }
+
+            zfree(copy);
+        }
+    } else if (op[0] == '-' && op[1] != '@') {
+        struct redisCommand *cmd = ACLLookupCommand(op+1);
+        if (cmd == NULL) {
+            errno = ENOENT;
+            return C_ERR;
+        }
+        ACLChangeSelectorPerm(selector,cmd,0);
+    } else if ((op[0] == '+' || op[0] == '-') && op[1] == '@') {
+        int bitval = op[0] == '+' ? 1 : 0;
+        if (ACLSetSelectorCommandBitsForCategory(selector,op+2,bitval) == C_ERR) {
+            errno = ENOENT;
+            return C_ERR;
+        }
+    } else {
+        errno = EINVAL;
+        return C_ERR;
+    }
+    return C_OK;
+}
+
+/* Set user properties according to the string "op". The following
+ * is a description of what different strings will do:
+ *
+ * on           Enable the user: it is possible to authenticate as this user.
+ * off          Disable the user: it's no longer possible to authenticate
+ *              with this user, however the already authenticated connections
+ *              will still work.
  * ><password>  Add this password to the list of valid password for the user.
  *              For example >mypass will add "mypass" to the list.
  *              This directive clears the "nopass" flag (see later).
@@ -849,6 +1202,17 @@ void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
  * reset        Performs the following actions: resetpass, resetkeys, off,
  *              -@all. The user returns to the same state it has immediately
  *              after its creation.
+ * (<options>)  Create a new selector with the options specified within the
+ *              parentheses and attach it to the user. Each option should be
+ *              space separated. The first character must be ( and the last
+ *              character must be ).
+ * clearselectors          Remove all of the currently attached selectors. 
+ *                         Note this does not change the "root" user permissions,
+ *                         which are the permissions directly applied onto the
+ *                         user (outside the parentheses).
+ * 
+ * Selector options can also be specified by this function, in which case
+ * they update the root selector for the user.
  *
  * The 'op' string must be null terminated. The 'oplen' argument should
  * specify the length of the 'op' string in case the caller requires to pass
@@ -888,34 +1252,6 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     } else if (!strcasecmp(op,"sanitize-payload")) {
         u->flags &= ~USER_FLAG_SANITIZE_PAYLOAD_SKIP;
         u->flags |= USER_FLAG_SANITIZE_PAYLOAD;
-    } else if (!strcasecmp(op,"allkeys") ||
-               !strcasecmp(op,"~*"))
-    {
-        u->flags |= USER_FLAG_ALLKEYS;
-        listEmpty(u->patterns);
-    } else if (!strcasecmp(op,"resetkeys")) {
-        u->flags &= ~USER_FLAG_ALLKEYS;
-        listEmpty(u->patterns);
-    } else if (!strcasecmp(op,"allchannels") ||
-               !strcasecmp(op,"&*"))
-    {
-        u->flags |= USER_FLAG_ALLCHANNELS;
-        listEmpty(u->channels);
-    } else if (!strcasecmp(op,"resetchannels")) {
-        u->flags &= ~USER_FLAG_ALLCHANNELS;
-        listEmpty(u->channels);
-    } else if (!strcasecmp(op,"allcommands") ||
-               !strcasecmp(op,"+@all"))
-    {
-        memset(u->allowed_commands,255,sizeof(u->allowed_commands));
-        u->flags |= USER_FLAG_ALLCOMMANDS;
-        ACLResetFirstArgs(u);
-    } else if (!strcasecmp(op,"nocommands") ||
-               !strcasecmp(op,"-@all"))
-    {
-        memset(u->allowed_commands,0,sizeof(u->allowed_commands));
-        u->flags &= ~USER_FLAG_ALLCOMMANDS;
-        ACLResetFirstArgs(u);
     } else if (!strcasecmp(op,"nopass")) {
         u->flags |= USER_FLAG_NOPASS;
         listEmpty(u->passwords);
@@ -960,123 +1296,39 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
             errno = ENODEV;
             return C_ERR;
         }
-    } else if (op[0] == '~') {
-        if (u->flags & USER_FLAG_ALLKEYS) {
-            errno = EEXIST;
+    } else if (op[0] == '(' && op[oplen - 1] == ')') {
+        aclSelector *selector = aclCreateSelectorFromOpSet(op, oplen);
+        if (!selector) {
+            /* No errorno set, propagate it from interior error. */
             return C_ERR;
         }
-        if (ACLStringHasSpaces(op+1,oplen-1)) {
-            errno = EINVAL;
-            return C_ERR;
-        }
-        sds newpat = sdsnewlen(op+1,oplen-1);
-        listNode *ln = listSearchKey(u->patterns,newpat);
-        /* Avoid re-adding the same key pattern multiple times. */
-        if (ln == NULL)
-            listAddNodeTail(u->patterns,newpat);
-        else
-            sdsfree(newpat);
-        u->flags &= ~USER_FLAG_ALLKEYS;
-    } else if (op[0] == '&') {
-        if (u->flags & USER_FLAG_ALLCHANNELS) {
-            errno = EISDIR;
-            return C_ERR;
-        }
-        if (ACLStringHasSpaces(op+1,oplen-1)) {
-            errno = EINVAL;
-            return C_ERR;
-        }
-        sds newpat = sdsnewlen(op+1,oplen-1);
-        listNode *ln = listSearchKey(u->channels,newpat);
-        /* Avoid re-adding the same channel pattern multiple times. */
-        if (ln == NULL)
-            listAddNodeTail(u->channels,newpat);
-        else
-            sdsfree(newpat);
-        u->flags &= ~USER_FLAG_ALLCHANNELS;
-    } else if (op[0] == '+' && op[1] != '@') {
-        if (strrchr(op,'|') == NULL) {
-            struct redisCommand *cmd = ACLLookupCommand(op+1);
-            if (cmd == NULL) {
-                errno = ENOENT;
-                return C_ERR;
-            }
-            ACLChangeCommandPerm(u,cmd,1);
-        } else {
-            /* Split the command and subcommand parts. */
-            char *copy = zstrdup(op+1);
-            char *sub = strrchr(copy,'|');
-            sub[0] = '\0';
-            sub++;
-
-            struct redisCommand *cmd = ACLLookupCommand(copy);
-
-            /* Check if the command exists. We can't check the
-             * subcommand to see if it is valid. */
-            if (cmd == NULL) {
-                zfree(copy);
-                errno = ENOENT;
-                return C_ERR;
-            }
-
-            /* The subcommand cannot be empty, so things like DEBUG|
-             * are syntax errors of course. */
-            if (strlen(sub) == 0) {
-                zfree(copy);
-                errno = EINVAL;
-                return C_ERR;
-            }
-
-            if (cmd->subcommands_dict) {
-                /* If user is trying to allow a valid subcommand we can just add its unique ID */
-                struct redisCommand *cmd = ACLLookupCommand(op+1);
-                if (cmd == NULL) {
-                    zfree(copy);
-                    errno = ENOENT;
-                    return C_ERR;
-                }
-                ACLChangeCommandPerm(u,cmd,1);
-            } else {
-                /* If user is trying to use the ACL mech to block SELECT except SELECT 0 or
-                 * block DEBUG except DEBUG OBJECT (DEBUG subcommands are not considered
-                 * subcommands for now) we use the allowed_firstargs mechanism. */
-                struct redisCommand *cmd = ACLLookupCommand(copy);
-                if (cmd == NULL) {
-                    zfree(copy);
-                    errno = ENOENT;
-                    return C_ERR;
-                }
-                /* Add the first-arg to the list of valid ones. */
-                ACLAddAllowedFirstArg(u,cmd->id,sub);
-            }
-
-            zfree(copy);
-        }
-    } else if (op[0] == '-' && op[1] != '@') {
-        struct redisCommand *cmd = ACLLookupCommand(op+1);
-        if (cmd == NULL) {
-            errno = ENOENT;
-            return C_ERR;
-        }
-        ACLChangeCommandPerm(u,cmd,0);
-    } else if ((op[0] == '+' || op[0] == '-') && op[1] == '@') {
-        int bitval = op[0] == '+' ? 1 : 0;
-        if (ACLSetUserCommandBitsForCategory(u,op+2,bitval) == C_ERR) {
-            errno = ENOENT;
-            return C_ERR;
+        listAddNodeTail(u->selectors, selector);
+        return C_OK;
+    } else if (!strcasecmp(op,"clearselectors")) {
+        listIter li;
+        listNode *ln;
+        listRewind(u->selectors,&li);
+        /* There has to be a root selector */
+        serverAssert(listNext(&li));
+        while((ln = listNext(&li))) {
+            listDelNode(u->selectors, ln);
         }
+        return C_OK;
     } else if (!strcasecmp(op,"reset")) {
         serverAssert(ACLSetUser(u,"resetpass",-1) == C_OK);
         serverAssert(ACLSetUser(u,"resetkeys",-1) == C_OK);
         serverAssert(ACLSetUser(u,"resetchannels",-1) == C_OK);
-        if (server.acl_pubsub_default & USER_FLAG_ALLCHANNELS)
+        if (server.acl_pubsub_default & SELECTOR_FLAG_ALLCHANNELS)
             serverAssert(ACLSetUser(u,"allchannels",-1) == C_OK);
         serverAssert(ACLSetUser(u,"off",-1) == C_OK);
         serverAssert(ACLSetUser(u,"sanitize-payload",-1) == C_OK);
+        serverAssert(ACLSetUser(u,"clearselectors",-1) == C_OK);
         serverAssert(ACLSetUser(u,"-@all",-1) == C_OK);
     } else {
-        errno = EINVAL;
-        return C_ERR;
+        aclSelector *selector = ACLUserGetRootSelector(u);
+        if (ACLSetSelector(selector, op, oplen) == C_ERR) {
+            return C_ERR;
+        }
     }
     return C_OK;
 }
@@ -1238,68 +1490,134 @@ user *ACLGetUserByName(const char *name, size_t namelen) {
     return myuser;
 }
 
-/* Check if the key can be accessed by the client according to
- * the ACLs associated with the specified user.
+/* =============================================================================
+ * ACL permission checks
+ * ==========================================================================*/
+
+/* Check if the key can be accessed by the selector.
  *
- * If the user can access the key, ACL_OK is returned, otherwise
+ * If the selector can access the key, ACL_OK is returned, otherwise
  * ACL_DENIED_KEY is returned. */
-int ACLCheckKey(const user *u, const char *key, int keylen) {
-    /* If there is no associated user, the connection can run anything. */
-    if (u == NULL) return ACL_OK;
-
-    /* The user can run any keys */
-    if (u->flags & USER_FLAG_ALLKEYS) return ACL_OK;
+static int ACLSelectorCheckKey(aclSelector *selector, const char *key, int keylen, int keyspec_flags) {
+    /* The selector can access any key */
+    if (selector->flags & SELECTOR_FLAG_ALLKEYS) return ACL_OK;
 
     listIter li;
     listNode *ln;
-    listRewind(u->patterns,&li);
+    listRewind(selector->patterns,&li);
+
+    int key_flags = 0;
+    if (keyspec_flags & CMD_KEY_ACCESS) key_flags |= ACL_READ_PERMISSION;
+    if (keyspec_flags & CMD_KEY_INSERT) key_flags |= ACL_WRITE_PERMISSION;
+    if (keyspec_flags & CMD_KEY_DELETE) key_flags |= ACL_WRITE_PERMISSION;
+    if (keyspec_flags & CMD_KEY_UPDATE) key_flags |= ACL_WRITE_PERMISSION;
 
     /* Test this key against every pattern. */
     while((ln = listNext(&li))) {
+        keyPattern *pattern = listNodeValue(ln);
+        if ((pattern->flags & key_flags) != key_flags)
+            continue;
+        size_t plen = sdslen(pattern->pattern);
+        if (stringmatchlen(pattern->pattern,plen,key,keylen,0))
+            return ACL_OK;
+    }
+    return ACL_DENIED_KEY;
+}
+
+/* Returns if a given command may possibly access channels. For this context,
+ * the unsubscribe commands do not have channels. */
+static int ACLDoesCommandHaveChannels(struct redisCommand *cmd) {
+    return (cmd->proc == publishCommand
+        || cmd->proc == subscribeCommand
+        || cmd->proc == psubscribeCommand
+        || cmd->proc == spublishCommand
+        || cmd->proc == ssubscribeCommand);
+}
+
+/* Checks a channel against a provide list of channels. */
+static int ACLCheckChannelAgainstList(list *reference, const char *channel, int channellen, int literal) {
+    listIter li;
+    listNode *ln;
+
+    listRewind(reference, &li);
+    while((ln = listNext(&li))) {
         sds pattern = listNodeValue(ln);
         size_t plen = sdslen(pattern);
-        if (stringmatchlen(pattern,plen,key,keylen,0))
+        if ((literal && !strcmp(pattern,channel)) || 
+            (!literal && stringmatchlen(pattern,plen,channel,channellen,0)))
+        {
             return ACL_OK;
+        }
     }
-    return ACL_DENIED_KEY;
+    return ACL_DENIED_CHANNEL;
 }
 
-/* Check if the command is ready to be executed according to the
- * ACLs associated with the specified user.
+/* Check if the pub/sub channels of the command can be executed
+ * according to the ACL channels associated with the specified selector.
+ * 
+ * idx and count are the index and count of channel arguments from the
+ * command. The literal argument controls whether the selector's ACL channels are 
+ * evaluated as literal values or matched as glob-like patterns.
  *
- * If the user can execute the command ACL_OK is returned, otherwise
- * ACL_DENIED_CMD or ACL_DENIED_KEY is returned: the first in case the
- * command cannot be executed because the user is not allowed to run such
- * command, the second if the command is denied because the user is trying
- * to access keys that are not among the specified patterns. */
-int ACLCheckCommandPerm(const user *u, struct redisCommand *cmd, robj **argv, int argc, int *keyidxptr) {
-    int ret;
-    uint64_t id = cmd->id;
+ * If the selector can execute the command ACL_OK is returned, otherwise 
+ * ACL_DENIED_CHANNEL. */
+static int ACLSelectorCheckPubsubArguments(aclSelector *s, robj **argv, int idx, int count, int literal, int *idxptr) {
+    for (int j = idx; j < idx+count; j++) {
+        if (ACLCheckChannelAgainstList(s->channels, argv[j]->ptr, sdslen(argv[j]->ptr), literal != ACL_OK)) {
+            if (idxptr) *idxptr = j;
+            return ACL_DENIED_CHANNEL;
+        }
+    }
 
-    /* If there is no associated user, the connection can run anything. */
-    if (u == NULL) return ACL_OK;
+    /* If we survived all the above checks, the selector can execute the
+     * command. */
+    return ACL_OK;
+}
 
-    /* Check if the user can execute this command or if the command
-     * doesn't need to be authenticated (hello, auth). */
-    if (!(u->flags & USER_FLAG_ALLCOMMANDS) && !(cmd->flags & CMD_NO_AUTH))
-    {
+/* To prevent duplicate calls to getKeysResult, a cache is maintained
+ * in between calls to the various selectors. */
+typedef struct {
+    int keys_init;
+    getKeysResult keys;
+} aclKeyResultCache;
+
+void initACLKeyResultCache(aclKeyResultCache *cache) {
+    cache->keys_init = 0;
+}
+
+void cleanupACLKeyResultCache(aclKeyResultCache *cache) {
+    if (cache->keys_init) getKeysFreeResult(&(cache->keys));
+}
+
+/* Check if the command is ready to be executed according to the
+ * ACLs associated with the specified selector.
+ *
+ * If the selector can execute the command ACL_OK is returned, otherwise
+ * ACL_DENIED_CMD, ACL_DENIED_KEY, or ACL_DENIED_CHANNEL is returned: the first in case the
+ * command cannot be executed because the selector is not allowed to run such
+ * command, the second and third if the command is denied because the selector is trying
+ * to access a key or channel that are not among the specified patterns. */
+static int ACLSelectorCheckCmd(aclSelector *selector, struct redisCommand *cmd, robj **argv, int argc, int *keyidxptr, aclKeyResultCache *cache) {
+    uint64_t id = cmd->id;
+    int ret;
+    if (!(selector->flags & SELECTOR_FLAG_ALLCOMMANDS) && !(cmd->flags & CMD_NO_AUTH)) {
         /* If the bit is not set we have to check further, in case the
          * command is allowed just with that specific first argument. */
-        if (ACLGetUserCommandBit(u,id) == 0) {
+        if (ACLGetSelectorCommandBit(selector,id) == 0) {
             /* Check if the first argument matches. */
             if (argc < 2 ||
-                u->allowed_firstargs == NULL ||
-                u->allowed_firstargs[id] == NULL)
+                selector->allowed_firstargs == NULL ||
+                selector->allowed_firstargs[id] == NULL)
             {
                 return ACL_DENIED_CMD;
             }
 
             long subid = 0;
             while (1) {
-                if (u->allowed_firstargs[id][subid] == NULL)
+                if (selector->allowed_firstargs[id][subid] == NULL)
                     return ACL_DENIED_CMD;
                 int idx = cmd->parent ? 2 : 1;
-                if (!strcasecmp(argv[idx]->ptr,u->allowed_firstargs[id][subid]))
+                if (!strcasecmp(argv[idx]->ptr,selector->allowed_firstargs[id][subid]))
                     break; /* First argument match found. Stop here. */
                 subid++;
             }
@@ -1307,101 +1625,218 @@ int ACLCheckCommandPerm(const user *u, struct redisCommand *cmd, robj **argv, in
     }
 
     /* Check if the user can execute commands explicitly touching the keys
-     * mentioned in the command arguments. Shard channels are treated as
-     * special keys for client library to rely on `COMMAND` command
-     * to discover the node to connect to. These don't need acl key check. */
-    if (!(u->flags & USER_FLAG_ALLKEYS) &&
-        !(cmd->flags & CMD_PUBSUB) &&
-        (cmd->getkeys_proc || cmd->key_specs_num))
-    {
-        getKeysResult result = GETKEYS_RESULT_INIT;
-        int numkeys = getKeysFromCommand(cmd,argv,argc,&result);
-        int *keyidx = result.keys;
-        for (int j = 0; j < numkeys; j++) {
-            int idx = keyidx[j];
-            ret = ACLCheckKey(u, argv[idx]->ptr, sdslen(argv[idx]->ptr));
+     * mentioned in the command arguments. */
+    if (!(selector->flags & SELECTOR_FLAG_ALLKEYS) && doesCommandHaveKeys(cmd)) {
+        if (!(cache->keys_init)) {
+            cache->keys = (getKeysResult) GETKEYS_RESULT_INIT;
+            getKeysFromCommandWithSpecs(cmd, argv, argc, GET_KEYSPEC_DEFAULT, &(cache->keys));
+            cache->keys_init = 1;
+        }
+        getKeysResult *result = &(cache->keys);
+        keyReference *resultidx = result->keys;
+        for (int j = 0; j < result->numkeys; j++) {
+            int idx = resultidx[j].pos;
+            ret = ACLSelectorCheckKey(selector, argv[idx]->ptr, sdslen(argv[idx]->ptr), resultidx[j].flags);
             if (ret != ACL_OK) {
-                if (keyidxptr) *keyidxptr = keyidx[j];
-                getKeysFreeResult(&result);
+                if (resultidx) *keyidxptr = resultidx[j].pos;
                 return ret;
             }
         }
-        getKeysFreeResult(&result);
     }
 
-    /* If we survived all the above checks, the user can execute the
-     * command. */
+    /* Check if the user can execute commands explicitly touching the channels
+     * mentioned in the command arguments */
+    if (!(selector->flags & SELECTOR_FLAG_ALLCHANNELS) && ACLDoesCommandHaveChannels(cmd)) {
+        if (cmd->proc == publishCommand || cmd->proc == spublishCommand) {
+            ret = ACLSelectorCheckPubsubArguments(selector,argv, 1, 1, 0, keyidxptr);
+        } else if (cmd->proc == subscribeCommand || cmd->proc == ssubscribeCommand) {
+            ret = ACLSelectorCheckPubsubArguments(selector, argv, 1, argc-1, 0, keyidxptr);
+        } else if (cmd->proc == psubscribeCommand) {
+            ret = ACLSelectorCheckPubsubArguments(selector, argv, 1, argc-1, 1, keyidxptr);
+        } else {
+            serverPanic("Encountered a command declared with channels but not handled");
+        }
+        if (ret != ACL_OK) {
+            /* keyidxptr is set by ACLSelectorCheckPubsubArguments */
+            return ret;
+        }
+    }
     return ACL_OK;
 }
 
-/* Check if the provided channel is whitelisted by the given allowed channels
- * list. Glob-style pattern matching is employed, unless the literal flag is
- * set. Returns ACL_OK if access is granted or ACL_DENIED_CHANNEL otherwise. */
-int ACLCheckPubsubChannelPerm(sds channel, list *allowed, int literal) {
+/* Check if the key can be accessed by the client according to
+ * the ACLs associated with the specified user.
+ *
+ * If the user can access the key, ACL_OK is returned, otherwise
+ * ACL_DENIED_KEY is returned. */
+int ACLUserCheckKeyPerm(user *u, const char *key, int keylen, int flags) {
     listIter li;
     listNode *ln;
-    size_t clen = sdslen(channel);
-    int match = 0;
 
-    listRewind(allowed,&li);
+    /* If there is no associated user, the connection can run anything. */
+    if (u == NULL) return ACL_OK;
+
+    /* Check all of the selectors */
+    listRewind(u->selectors,&li);
     while((ln = listNext(&li))) {
-        sds pattern = listNodeValue(ln);
-        size_t plen = sdslen(pattern);
+        aclSelector *s = (aclSelector *) listNodeValue(ln);
+        if (ACLSelectorCheckKey(s, key, keylen, flags) == ACL_OK) {
+            return ACL_OK;
+        }
+    }
+    return ACL_DENIED_KEY;
+}
 
-        if ((literal && !sdscmp(pattern,channel)) || 
-            (!literal && stringmatchlen(pattern,plen,channel,clen,0)))
-        {
-            match = 1;
-            break;
+/* Check if the channel can be accessed by the client according to
+ * the ACLs associated with the specified user.
+ *
+ * If the user can access the key, ACL_OK is returned, otherwise
+ * ACL_DENIED_CHANNEL is returned. */
+int ACLUserCheckChannelPerm(user *u, sds channel, int literal) {
+    listIter li;
+    listNode *ln;
+
+    /* If there is no associated user, the connection can run anything. */
+    if (u == NULL) return ACL_OK;
+
+    /* Check all of the selectors */
+    listRewind(u->selectors,&li);
+    while((ln = listNext(&li))) {
+        aclSelector *s = (aclSelector *) listNodeValue(ln);
+        /* The selector can run any keys */
+        if (s->flags & SELECTOR_FLAG_ALLCHANNELS) return ACL_OK;
+
+        /* Otherwise, loop over the selectors list and check each channel */
+        if (ACLCheckChannelAgainstList(s->channels, channel, sdslen(channel), literal) == ACL_OK) {
+            return ACL_OK;
         }
     }
-    if (!match) {
-        return ACL_DENIED_CHANNEL;
+    return ACL_DENIED_CHANNEL;
+}
+
+/* Lower level API that checks if a specified user is able to execute a given command. */
+int ACLCheckAllUserCommandPerm(user *u, struct redisCommand *cmd, robj **argv, int argc, int *idxptr) {
+    listIter li;
+    listNode *ln;
+
+    /* If there is no associated user, the connection can run anything. */
+    if (u == NULL) return ACL_OK;
+
+    /* We have to pick a single error to log, the logic for picking is as follows:
+     * 1) If no selector can execute the command, return the command.
+     * 2) Return the last key or channel that no selector could match. */
+    int relevant_error = ACL_DENIED_CMD;
+    int local_idxptr = 0, last_idx = 0;
+
+    /* For multiple selectors, we cache the key result in between selector
+     * calls to prevent duplicate lookups. */
+    aclKeyResultCache cache;
+    initACLKeyResultCache(&cache);
+
+    /* Check each selector sequentially */
+    listRewind(u->selectors,&li);
+    while((ln = listNext(&li))) {
+        aclSelector *s = (aclSelector *) listNodeValue(ln);
+        int acl_retval = ACLSelectorCheckCmd(s, cmd, argv, argc, &local_idxptr, &cache);
+        if (acl_retval == ACL_OK) {
+            cleanupACLKeyResultCache(&cache);
+            return ACL_OK;
+        }
+        if (acl_retval > relevant_error ||
+            (acl_retval == relevant_error && local_idxptr > last_idx))
+        {
+            relevant_error = acl_retval;
+            last_idx = local_idxptr;
+        }
     }
-    return ACL_OK;
+
+    *idxptr = last_idx;
+    cleanupACLKeyResultCache(&cache);
+    return relevant_error;
+}
+
+/* High level API for checking if a client can execute the queued up command */
+int ACLCheckAllPerm(client *c, int *idxptr) {
+    return ACLCheckAllUserCommandPerm(c->user, c->cmd, c->argv, c->argc, idxptr);
 }
 
 /* Check if the user's existing pub/sub clients violate the ACL pub/sub
  * permissions specified via the upcoming argument, and kill them if so. */
-void ACLKillPubsubClientsIfNeeded(user *u, list *upcoming) {
+void ACLKillPubsubClientsIfNeeded(user *new, user *original) {
     listIter li, lpi;
     listNode *ln, *lpn;
     robj *o;
     int kill = 0;
     
-    /* Nothing to kill when the upcoming are a literal super set of the original
+    /* First optimization is we check if any selector has all channel
      * permissions. */
-    listRewind(u->channels,&li);
-    while (!kill && ((ln = listNext(&li)) != NULL)) {
-        sds pattern = listNodeValue(ln);
-        kill = (ACLCheckPubsubChannelPerm(pattern,upcoming,1) ==
-                ACL_DENIED_CHANNEL);
+    listRewind(new->selectors,&li);
+    while((ln = listNext(&li))) {
+        aclSelector *s = (aclSelector *) listNodeValue(ln);
+        if (s->flags & SELECTOR_FLAG_ALLCHANNELS) return;
+    }
+
+    /* Second optimization is to check if the new list of channels
+     * is a strict superset of the original. This is done by
+     * created an "upcoming" list of all channels that are in
+     * the new user and checking each of the existing channels
+     * against it.  */
+    list *upcoming = listCreate();
+    listRewind(new->selectors,&li);
+    while((ln = listNext(&li))) {
+        aclSelector *s = (aclSelector *) listNodeValue(ln);
+        listRewind(s->channels, &lpi);
+        while((lpn = listNext(&lpi))) {
+            listAddNodeTail(upcoming, listNodeValue(lpn));
+        }
+    }
+
+    int match = 1;
+    listRewind(original->selectors,&li);
+    while((ln = listNext(&li)) && match) {
+        aclSelector *s = (aclSelector *) listNodeValue(ln);
+        listRewind(s->channels, &lpi);
+        while((lpn = listNext(&lpi)) && match) {
+            if (!listSearchKey(upcoming, listNodeValue(lpn))) {
+                match = 0;
+                break;
+            }
+        }
     }
-    if (!kill) return;
 
-    /* Scan all connected clients to find the user's pub/subs. */
+    if (match) {
+        /* All channels were matched, no need to kill clients. */
+        listRelease(upcoming);
+        return;
+    }
+    
+    /* Permissions have changed, so we need to iterate through all
+     * the clients and disconnect those that are no longer valid.
+     * Scan all connected clients to find the user's pub/subs. */
     listRewind(server.clients,&li);
     while ((ln = listNext(&li)) != NULL) {
         client *c = listNodeValue(ln);
         kill = 0;
 
-        if (c->user == u && getClientType(c) == CLIENT_TYPE_PUBSUB) {
+        if (c->user == original && getClientType(c) == CLIENT_TYPE_PUBSUB) {
             /* Check for pattern violations. */
             listRewind(c->pubsub_patterns,&lpi);
             while (!kill && ((lpn = listNext(&lpi)) != NULL)) {
+
                 o = lpn->value;
-                kill = (ACLCheckPubsubChannelPerm(o->ptr,upcoming,1) == 
-                        ACL_DENIED_CHANNEL);
+                int res = ACLCheckChannelAgainstList(upcoming, o->ptr, sdslen(o->ptr), 1);
+                kill = (res == ACL_DENIED_CHANNEL);
             }
             /* Check for channel violations. */
             if (!kill) {
                 /* Check for global channels violation. */
                 dictIterator *di = dictGetIterator(c->pubsub_channels);
+
                 dictEntry *de;                
                 while (!kill && ((de = dictNext(di)) != NULL)) {
                     o = dictGetKey(de);
-                    kill = (ACLCheckPubsubChannelPerm(o->ptr,upcoming,0) ==
-                            ACL_DENIED_CHANNEL);
+                    int res = ACLCheckChannelAgainstList(upcoming, o->ptr, sdslen(o->ptr), 0);
+                    kill = (res == ACL_DENIED_CHANNEL);
                 }
                 dictReleaseIterator(di);
 
@@ -1409,8 +1844,8 @@ void ACLKillPubsubClientsIfNeeded(user *u, list *upcoming) {
                 di = dictGetIterator(c->pubsubshard_channels);
                 while (!kill && ((de = dictNext(di)) != NULL)) {
                     o = dictGetKey(de);
-                    kill = (ACLCheckPubsubChannelPerm(o->ptr,upcoming,0) ==
-                            ACL_DENIED_CHANNEL);
+                    int res = ACLCheckChannelAgainstList(upcoming, o->ptr, sdslen(o->ptr), 0);
+                    kill = (res == ACL_DENIED_CHANNEL);
                 }
 
                 dictReleaseIterator(di);
@@ -1422,63 +1857,66 @@ void ACLKillPubsubClientsIfNeeded(user *u, list *upcoming) {
             }
         }
     }
+    listRelease(upcoming);
 }
 
-/* Check if the pub/sub channels of the command, that's ready to be executed
- * according to the ACLs channels associated with the specified user.
+/* =============================================================================
+ * ACL loading / saving functions
+ * ==========================================================================*/
+
+
+/* Selector definitions should be sent as a single argument, however
+ * we will be lenient and try to find selector definitions spread 
+ * across multiple arguments since it makes for a simpler user experience
+ * for ACL SETUSER as well as when loading from conf files. 
  * 
- * idx and count are the index and count of channel arguments from the
- * command. The literal argument controls whether the user's ACL channels are 
- * evaluated as literal values or matched as glob-like patterns.
- *
- * If the user can execute the command ACL_OK is returned, otherwise 
- * ACL_DENIED_CHANNEL. */
-int ACLCheckPubsubPerm(const user *u, robj **argv, int idx, int count, int literal, int *idxptr) {
-    /* If there is no associated user, the connection can run anything. */
-    if (u == NULL) return ACL_OK;
+ * This function takes in an array of ACL operators, excluding the username,
+ * and merges selector operations that are spread across multiple arguments. The return
+ * value is a new SDS array, with length set to the passed in merged_argc. Arguments 
+ * that are untouched are still duplicated. If there is an unmatched parenthesis, NULL 
+ * is returned and invalid_idx is set to the argument with the start of the opening
+ * parenthesis. */
+sds *ACLMergeSelectorArguments(sds *argv, int argc, int *merged_argc, int *invalid_idx) {
+    *merged_argc = 0;
+    int open_bracket_start = -1;
+
+    sds *acl_args = (sds *) zmalloc(sizeof(sds) * argc);
+
+    sds selector = NULL;
+    for (int j = 0; j < argc; j++) {
+        char *op = argv[j];
+
+        if (op[0] == '(' && op[sdslen(op) - 1] != ')') {
+            selector = sdsdup(argv[j]);
+            open_bracket_start = j;
+            continue;
+        }
 
-    /* Check if the user can access the channels mentioned in the command's
-     * arguments. */
-    if (!(u->flags & USER_FLAG_ALLCHANNELS)) {
-        for (int j = idx; j < idx+count; j++) {
-            if (ACLCheckPubsubChannelPerm(argv[j]->ptr,u->channels,literal)
-                != ACL_OK) {
-                if (idxptr) *idxptr = j;
-                return ACL_DENIED_CHANNEL;
+        if (open_bracket_start != -1) {
+            selector = sdscatfmt(selector, " %s", op);
+            if (op[sdslen(op) - 1] == ')') {
+                open_bracket_start = -1;
+                acl_args[*merged_argc] = selector;                        
+                (*merged_argc)++;
             }
+            continue;
         }
-    }
-
-    /* If we survived all the above checks, the user can execute the
-     * command. */
-    return ACL_OK;
 
-}
+        acl_args[*merged_argc] = sdsdup(argv[j]);
+        (*merged_argc)++;
+    }
 
-/* Check whether the command is ready to be executed by ACLCheckCommandPerm.
- * If check passes, then check whether pub/sub channels of the command is
- * ready to be executed by ACLCheckPubsubPerm */
-int ACLCheckAllUserCommandPerm(const user *u, struct redisCommand *cmd, robj **argv, int argc, int *idxptr) {
-    int acl_retval = ACLCheckCommandPerm(u,cmd,argv,argc,idxptr);
-    if (acl_retval != ACL_OK)
-        return acl_retval;
-    if (cmd->proc == publishCommand || cmd->proc == spublishCommand)
-        acl_retval = ACLCheckPubsubPerm(u,argv,1,1,0,idxptr);
-    else if (cmd->proc == subscribeCommand || cmd->proc == ssubscribeCommand)
-        acl_retval = ACLCheckPubsubPerm(u,argv,1,argc-1,0,idxptr);
-    else if (cmd->proc == psubscribeCommand)
-        acl_retval = ACLCheckPubsubPerm(u,argv,1,argc-1,1,idxptr);
-    return acl_retval;
-}
+    if (open_bracket_start != -1) {
+        for (int i = 0; i < *merged_argc; i++) sdsfree(acl_args[i]);
+        zfree(acl_args);
+        sdsfree(selector);
+        if (invalid_idx) *invalid_idx = open_bracket_start;
+        return NULL;
+    }
 
-int ACLCheckAllPerm(client *c, int *idxptr) {
-    return ACLCheckAllUserCommandPerm(c->user, c->cmd, c->argv, c->argc, idxptr);
+    return acl_args;
 }
 
-/* =============================================================================
- * ACL loading / saving functions
- * ==========================================================================*/
-
 /* Given an argument vector describing a user in the form:
  *
  *      user <username> ... ACL rules and flags ...
@@ -1513,22 +1951,35 @@ int ACLAppendUserForLoading(sds *argv, int argc, int *argc_err) {
      * are actually valid. */
     user *fakeuser = ACLCreateUnlinkedUser();
 
-    for (int j = 2; j < argc; j++) {
-        if (ACLSetUser(fakeuser,argv[j],sdslen(argv[j])) == C_ERR) {
+    /* Merged selectors before trying to process */
+    int merged_argc;
+    sds *acl_args = ACLMergeSelectorArguments(argv + 2, argc - 2, &merged_argc, argc_err);
+
+    if (!acl_args) {
+        return C_ERR;
+    }
+
+    for (int j = 0; j < merged_argc; j++) {
+        if (ACLSetUser(fakeuser,acl_args[j],sdslen(acl_args[j])) == C_ERR) {
             if (errno != ENOENT) {
                 ACLFreeUser(fakeuser);
                 if (argc_err) *argc_err = j;
+                for (int i = 0; i < merged_argc; i++) sdsfree(acl_args[i]);
+                zfree(acl_args);
                 return C_ERR;
             }
         }
     }
 
     /* Rules look valid, let's append the user to the list. */
-    sds *copy = zmalloc(sizeof(sds)*argc);
-    for (int j = 1; j < argc; j++) copy[j-1] = sdsdup(argv[j]);
-    copy[argc-1] = NULL;
+    sds *copy = zmalloc(sizeof(sds)*(merged_argc + 2));
+    copy[0] = sdsdup(argv[1]);
+    for (int j = 0; j < merged_argc; j++) copy[j+1] = sdsdup(acl_args[j]);
+    copy[merged_argc + 1] = NULL;
     listAddNodeTail(UsersToLoad,copy);
     ACLFreeUser(fakeuser);
+    for (int i = 0; i < merged_argc; i++) sdsfree(acl_args[i]);
+    zfree(acl_args);
     return C_OK;
 }
 
@@ -1690,10 +2141,18 @@ sds ACLLoadFromFile(const char *filename) {
          * be cleanly applied to the user. If any option fails
          * to apply, the other values won't be applied since
          * all the pending changes will get dropped. */
+        int merged_argc;
+        sds *acl_args = ACLMergeSelectorArguments(argv + 2, argc - 2, &merged_argc, NULL);
+        if (!acl_args) {
+            errors = sdscatprintf(errors,
+                    "%s:%d: Unmatched parenthesis in selector definition.",
+                    server.acl_filename, linenum);
+        }
+
         int j;
-        for (j = 2; j < argc; j++) {
-            argv[j] = sdstrim(argv[j],"\t\r\n");
-            if (ACLSetUser(u,argv[j],sdslen(argv[j])) != C_OK) {
+        for (j = 0; j < merged_argc; j++) {
+            acl_args[j] = sdstrim(acl_args[j],"\t\r\n");
+            if (ACLSetUser(u,acl_args[j],sdslen(acl_args[j])) != C_OK) {
                 const char *errmsg = ACLSetUserStringError();
                 errors = sdscatprintf(errors,
                          "%s:%d: %s. ",
@@ -1702,6 +2161,9 @@ sds ACLLoadFromFile(const char *filename) {
             }
         }
 
+        for (int i = 0; i < merged_argc; i++) sdsfree(acl_args[i]);
+        zfree(acl_args);
+
         /* Apply the rule to the new users set only if so far there
          * are no errors, otherwise it's useless since we are going
          * to discard the new users set anyway. */
@@ -1965,6 +2427,53 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username
  * ACL related commands
  * ==========================================================================*/
 
+/* Add the formatted response from a single selector to the ACL GETUSER
+ * response. This function returns the number of fields added. 
+ * 
+ * Setting verbose to 1 means that the full qualifier for key and channel
+ * permissions are shown.
+ */
+int aclAddReplySelectorDescription(client *c, aclSelector *s) {
+    listIter li;
+    listNode *ln;
+
+    /* Commands */
+    addReplyBulkCString(c,"commands");
+    sds cmddescr = ACLDescribeSelectorCommandRules(s);
+    addReplyBulkSds(c,cmddescr);
+    
+    /* Key patterns */
+    addReplyBulkCString(c,"keys");
+    if (s->flags & SELECTOR_FLAG_ALLKEYS) {
+        addReplyBulkCBuffer(c,"~*",2);
+    } else {
+        sds dsl = sdsempty();
+        listRewind(s->patterns,&li);
+        while((ln = listNext(&li))) {
+            keyPattern *thispat = (keyPattern *) listNodeValue(ln);
+            if (ln != listFirst(s->patterns)) dsl = sdscat(dsl, " ");
+            dsl = sdsCatPatternString(dsl, thispat);
+        }
+        addReplyBulkSds(c, dsl);
+    }
+
+    /* Pub/sub patterns */
+    addReplyBulkCString(c,"channels");
+    if (s->flags & SELECTOR_FLAG_ALLCHANNELS) {
+        addReplyBulkCBuffer(c,"&*",2);
+    } else {
+        sds dsl = sdsempty();
+        listRewind(s->channels,&li);
+        while((ln = listNext(&li))) {
+            sds thispat = listNodeValue(ln);
+            if (ln != listFirst(s->channels)) dsl = sdscat(dsl, " ");
+            dsl = sdscatfmt(dsl, "&%S", thispat);
+        }
+        addReplyBulkSds(c, dsl);
+    }
+    return 3;
+}
+
 /* ACL -- show and modify the configuration of ACL users.
  * ACL HELP
  * ACL LOAD
@@ -1996,6 +2505,19 @@ void aclCommand(client *c) {
             return;
         }
 
+        int merged_argc = 0, invalid_idx = 0;
+        sds *temp_argv = zmalloc(c->argc * sizeof(sds *));
+        for (int i = 3; i < c->argc; i++) temp_argv[i-3] = c->argv[i]->ptr;
+        sds *acl_args = ACLMergeSelectorArguments(temp_argv, c->argc - 3, &merged_argc, &invalid_idx);
+        zfree(temp_argv);
+
+        if (!acl_args) {
+            addReplyErrorFormat(c,
+                "Unmatched parenthesis in acl selector starting "
+                "at '%s'.", (char *) c->argv[invalid_idx]->ptr);
+            return;
+        }
+
         /* Create a temporary user to validate and stage all changes against
          * before applying to an existing user or creating a new user. If all
          * arguments are valid the user parameters will all be applied together.
@@ -2004,29 +2526,30 @@ void aclCommand(client *c) {
         user *u = ACLGetUserByName(username,sdslen(username));
         if (u) ACLCopyUser(tempu, u);
 
-        for (int j = 3; j < c->argc; j++) {
-            if (ACLSetUser(tempu,c->argv[j]->ptr,sdslen(c->argv[j]->ptr)) != C_OK) {
+        for (int j = 0; j < merged_argc; j++) {
+            if (ACLSetUser(tempu,acl_args[j],sdslen(acl_args[j])) != C_OK) {
                 const char *errmsg = ACLSetUserStringError();
                 addReplyErrorFormat(c,
                     "Error in ACL SETUSER modifier '%s': %s",
-                    (char*)c->argv[j]->ptr, errmsg);
-
-                ACLFreeUser(tempu);
-                return;
+                    (char*)acl_args[j], errmsg);
+                goto setuser_cleanup;
             }
         }
 
         /* Existing pub/sub clients authenticated with the user may need to be
          * disconnected if (some of) their channel permissions were revoked. */
-        if (u && !(tempu->flags & USER_FLAG_ALLCHANNELS))
-            ACLKillPubsubClientsIfNeeded(u,tempu->channels);
+        if (u) ACLKillPubsubClientsIfNeeded(tempu, u);
 
         /* Overwrite the user with the temporary user we modified above. */
         if (!u) u = ACLCreateUser(username,sdslen(username));
         serverAssert(u != NULL);
         ACLCopyUser(u, tempu);
-        ACLFreeUser(tempu);
         addReply(c,shared.ok);
+setuser_cleanup:
+        ACLFreeUser(tempu);
+        for (int i = 0; i < merged_argc; i++) sdsfree(acl_args[i]);
+        zfree(acl_args);
+        return;
     } else if (!strcasecmp(sub,"deluser") && c->argc >= 3) {
         int deleted = 0;
         for (int j = 2; j < c->argc; j++) {
@@ -2056,7 +2579,8 @@ void aclCommand(client *c) {
             return;
         }
 
-        addReplyMapLen(c,5);
+        void *ufields = addReplyDeferredLen(c);
+        int fields = 3;
 
         /* Flags */
         addReplyBulkCString(c,"flags");
@@ -2080,43 +2604,20 @@ void aclCommand(client *c) {
             sds thispass = listNodeValue(ln);
             addReplyBulkCBuffer(c,thispass,sdslen(thispass));
         }
+        /* Include the root selector at the top level for backwards compatibility */
+        fields += aclAddReplySelectorDescription(c, ACLUserGetRootSelector(u));
 
-        /* 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));
-            }
-        }
-
-        /* Pub/sub patterns */
-        addReplyBulkCString(c,"channels");
-        if (u->flags & USER_FLAG_ALLCHANNELS) {
-            addReplyArrayLen(c,1);
-            addReplyBulkCBuffer(c,"*",1);
-        } else {
-            addReplyArrayLen(c,listLength(u->channels));
-            listIter li;
-            listNode *ln;
-            listRewind(u->channels,&li);
-            while((ln = listNext(&li))) {
-                sds thispat = listNodeValue(ln);
-                addReplyBulkCBuffer(c,thispat,sdslen(thispat));
-            }
-        }
+        /* Describe all of the selectors on this user, including duplicating the root selector */
+        addReplyBulkCString(c,"selectors");
+        addReplyArrayLen(c, listLength(u->selectors) - 1);
+        listRewind(u->selectors,&li);
+        serverAssert(listNext(&li));
+        while((ln = listNext(&li))) {
+            void *slen = addReplyDeferredLen(c);
+            int sfields = aclAddReplySelectorDescription(c, (aclSelector *)listNodeValue(ln));
+            setDeferredMapLen(c, slen, sfields);
+        } 
+        setDeferredMapLen(c, ufields, fields);
     } else if ((!strcasecmp(sub,"list") || !strcasecmp(sub,"users")) &&
                c->argc == 2)
     {
@@ -2281,6 +2782,40 @@ void aclCommand(client *c) {
             addReplyBulkCString(c,"client-info");
             addReplyBulkCBuffer(c,le->cinfo,sdslen(le->cinfo));
         }
+    } else if (!strcasecmp(sub,"dryrun") && c->argc >= 4) {
+        struct redisCommand *cmd;
+        user *u = ACLGetUserByName(c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
+        if (u == NULL) {
+            addReplyErrorFormat(c, "User '%s' not found", (char *)c->argv[2]->ptr);
+            return;
+        }
+
+        if ((cmd = lookupCommand(c->argv + 3, c->argc - 3)) == NULL) {
+            addReplyErrorFormat(c, "Command '%s' not found", (char *)c->argv[3]->ptr);
+            return;
+        }
+
+        int idx;
+        int result = ACLCheckAllUserCommandPerm(u, cmd, c->argv + 3, c->argc - 3, &idx);
+        if (result != ACL_OK) {
+            sds err = sdsempty();
+            if (result == ACL_DENIED_CMD) {
+                err = sdscatfmt(err, "This user has no permissions to run "
+                    "the '%s' command or its subcommand", c->cmd->name);
+            } else if (result == ACL_DENIED_KEY) {
+                err = sdscatfmt(err, "This user has no permissions to access "
+                    "the '%s' key", c->argv[idx + 3]->ptr);
+            } else if (result == ACL_DENIED_CHANNEL) {
+                err = sdscatfmt(err, "This user has no permissions to access "
+                    "the '%s' channel", c->argv[idx + 3]->ptr);
+            } else {
+                serverPanic("Invalid permission result");
+            }
+            addReplyBulkSds(c, err);
+            return;
+        }
+
+        addReply(c,shared.ok);
     } else if (c->argc == 2 && !strcasecmp(sub,"help")) {
         const char *help[] = {
 "CAT [<category>]",
@@ -2288,6 +2823,8 @@ void aclCommand(client *c) {
 "    when no category is specified.",
 "DELUSER <username> [<username> ...]",
 "    Delete a list of users.",
+"DRYRUN <username> <command> [<arg> ...]",
+"    Returns whether the user can execute the given command without executing the command.",
 "GETUSER <username>",
 "    Get the user's details.",
 "GENPASS [<bits>]",
diff --git a/src/cluster.c b/src/cluster.c
index adc1629c8..df59f9ae7 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -6423,7 +6423,8 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
     for (i = 0; i < ms->count; i++) {
         struct redisCommand *mcmd;
         robj **margv;
-        int margc, *keyindex, numkeys, j;
+        int margc, numkeys, j;
+        keyReference *keyindex;
 
         mcmd = ms->commands[i].cmd;
         margc = ms->commands[i].argc;
@@ -6434,7 +6435,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
         keyindex = result.keys;
 
         for (j = 0; j < numkeys; j++) {
-            robj *thiskey = margv[keyindex[j]];
+            robj *thiskey = margv[keyindex[j].pos];
             int thisslot = keyHashSlot((char*)thiskey->ptr,
                                        sdslen(thiskey->ptr));
 
diff --git a/src/commands.c b/src/commands.c
index e97e6a486..d854c13eb 100644
--- a/src/commands.c
+++ b/src/commands.c
@@ -3812,6 +3812,22 @@ struct redisCommandArg ACL_DELUSER_Args[] = {
 {0}
 };
 
+/********** ACL DRYRUN ********************/
+
+/* ACL DRYRUN history */
+#define ACL_DRYRUN_History NULL
+
+/* ACL DRYRUN tips */
+#define ACL_DRYRUN_tips NULL
+
+/* ACL DRYRUN argument table */
+struct redisCommandArg ACL_DRYRUN_Args[] = {
+{"username",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
+{"command",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
+{"arg",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL|CMD_ARG_MULTIPLE},
+{0}
+};
+
 /********** ACL GENPASS ********************/
 
 /* ACL GENPASS history */
@@ -3901,6 +3917,7 @@ struct redisCommandArg ACL_LOG_Args[] = {
 /* ACL SETUSER history */
 commandHistory ACL_SETUSER_History[] = {
 {"6.2.0","Added Pub/Sub channel patterns."},
+{"7.0.0","Added selectors and key based permissions."},
 {0}
 };
 
@@ -3934,6 +3951,7 @@ struct redisCommandArg ACL_SETUSER_Args[] = {
 struct redisCommand ACL_Subcommands[] = {
 {"cat","List the ACL categories or the commands inside a category","O(1) since the categories and commands are a fixed set.","6.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,ACL_CAT_History,ACL_CAT_tips,aclCommand,-2,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,0,.args=ACL_CAT_Args},
 {"deluser","Remove the specified ACL users and the associated rules","O(1) amortized time considering the typical user.","6.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,ACL_DELUSER_History,ACL_DELUSER_tips,aclCommand,-3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,0,.args=ACL_DELUSER_Args},
+{"dryrun","Returns whether the user can execute the given command without executing the command.","O(1).","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,ACL_DRYRUN_History,ACL_DRYRUN_tips,aclCommand,-4,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,0,.args=ACL_DRYRUN_Args},
 {"genpass","Generate a pseudorandom secure password to use for ACL users","O(1)","6.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,ACL_GENPASS_History,ACL_GENPASS_tips,aclCommand,-2,CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,0,.args=ACL_GENPASS_Args},
 {"getuser","Get the rules for a specific ACL user","O(N). Where N is the number of password, command and pattern rules that the user has.","6.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,ACL_GETUSER_History,ACL_GETUSER_tips,aclCommand,3,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_SENTINEL,0,.args=ACL_GETUSER_Args},
 {"help","Show helpful text about the different subcommands","O(1)","6.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SERVER,ACL_HELP_History,ACL_HELP_tips,aclCommand,2,CMD_LOADING|CMD_STALE|CMD_SENTINEL,0},
@@ -6830,8 +6848,8 @@ struct redisCommand redisCommandTable[] = {
 {"renamenx","Rename a key, only if the new key does not exist","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,RENAMENX_History,RENAMENX_tips,renamenxCommand,3,CMD_WRITE|CMD_FAST,ACL_CATEGORY_KEYSPACE,{{CMD_KEY_RW|CMD_KEY_ACCESS|CMD_KEY_DELETE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}},{CMD_KEY_OW|CMD_KEY_INSERT,KSPEC_BS_INDEX,.bs.index={2},KSPEC_FK_RANGE,.fk.range={0,1,0}}},.args=RENAMENX_Args},
 {"restore","Create a key using the provided serialized value, previously obtained using DUMP.","O(1) to create the new key and additional O(N*M) to reconstruct the serialized value, where N is the number of Redis objects composing the value and M their average size. For small string values the time complexity is thus O(1)+O(1*M) where M is small, so simply O(1). However for sorted set values the complexity is O(N*M*log(N)) because inserting values into sorted sets is O(log(N)).","2.6.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,RESTORE_History,RESTORE_tips,restoreCommand,-4,CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_KEYSPACE|ACL_CATEGORY_DANGEROUS,{{CMD_KEY_OW|CMD_KEY_UPDATE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}}},.args=RESTORE_Args},
 {"scan","Incrementally iterate the keys space","O(1) for every call. O(N) for a complete iteration, including enough command calls for the cursor to return back to 0. N is the number of elements inside the collection.","2.8.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,SCAN_History,SCAN_tips,scanCommand,-2,CMD_READONLY,ACL_CATEGORY_KEYSPACE,.args=SCAN_Args},
-{"sort","Sort the elements in a list, set or sorted set","O(N+M*log(M)) where N is the number of elements in the list or set to sort, and M the number of returned elements. When the elements are not sorted, complexity is O(N).","1.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,SORT_History,SORT_tips,sortCommand,-2,CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_SET|ACL_CATEGORY_SORTEDSET|ACL_CATEGORY_LIST|ACL_CATEGORY_DANGEROUS,{{CMD_KEY_RO|CMD_KEY_ACCESS|CMD_KEY_INCOMPLETE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}},{CMD_KEY_OW|CMD_KEY_UPDATE|CMD_KEY_INCOMPLETE,KSPEC_BS_UNKNOWN,{{0}},KSPEC_FK_UNKNOWN,{{0}}}},sortGetKeys,.args=SORT_Args},
-{"sort_ro","Sort the elements in a list, set or sorted set. Read-only variant of SORT.","O(N+M*log(M)) where N is the number of elements in the list or set to sort, and M the number of returned elements. When the elements are not sorted, complexity is O(N).","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,SORT_RO_History,SORT_RO_tips,sortroCommand,-2,CMD_READONLY,ACL_CATEGORY_SET|ACL_CATEGORY_SORTEDSET|ACL_CATEGORY_LIST|ACL_CATEGORY_DANGEROUS,{{CMD_KEY_RO|CMD_KEY_ACCESS|CMD_KEY_INCOMPLETE,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}}},.args=SORT_RO_Args},
+{"sort","Sort the elements in a list, set or sorted set","O(N+M*log(M)) where N is the number of elements in the list or set to sort, and M the number of returned elements. When the elements are not sorted, complexity is O(N).","1.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,SORT_History,SORT_tips,sortCommand,-2,CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_SET|ACL_CATEGORY_SORTEDSET|ACL_CATEGORY_LIST|ACL_CATEGORY_DANGEROUS,{{CMD_KEY_RO|CMD_KEY_ACCESS,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}},{CMD_KEY_RO|CMD_KEY_ACCESS|CMD_KEY_INCOMPLETE,KSPEC_BS_UNKNOWN,{{0}},KSPEC_FK_UNKNOWN,{{0}}},{CMD_KEY_OW|CMD_KEY_UPDATE|CMD_KEY_INCOMPLETE,KSPEC_BS_UNKNOWN,{{0}},KSPEC_FK_UNKNOWN,{{0}}}},sortGetKeys,.args=SORT_Args},
+{"sort_ro","Sort the elements in a list, set or sorted set. Read-only variant of SORT.","O(N+M*log(M)) where N is the number of elements in the list or set to sort, and M the number of returned elements. When the elements are not sorted, complexity is O(N).","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,SORT_RO_History,SORT_RO_tips,sortroCommand,-2,CMD_READONLY,ACL_CATEGORY_SET|ACL_CATEGORY_SORTEDSET|ACL_CATEGORY_LIST|ACL_CATEGORY_DANGEROUS,{{CMD_KEY_RO|CMD_KEY_ACCESS,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}},{CMD_KEY_RO|CMD_KEY_ACCESS|CMD_KEY_INCOMPLETE,KSPEC_BS_UNKNOWN,{{0}},KSPEC_FK_UNKNOWN,{{0}}}},sortROGetKeys,.args=SORT_RO_Args},
 {"touch","Alters the last access time of a key(s). Returns the number of existing keys specified.","O(N) where N is the number of keys that will be touched.","3.2.1",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,TOUCH_History,TOUCH_tips,touchCommand,-2,CMD_READONLY|CMD_FAST,ACL_CATEGORY_KEYSPACE,{{CMD_KEY_RO,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={-1,1,0}}},.args=TOUCH_Args},
 {"ttl","Get the time to live for a key in seconds","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,TTL_History,TTL_tips,ttlCommand,2,CMD_READONLY|CMD_FAST,ACL_CATEGORY_KEYSPACE,{{CMD_KEY_RO|CMD_KEY_ACCESS,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}}},.args=TTL_Args},
 {"type","Determine the type stored at key","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_GENERIC,TYPE_History,TYPE_tips,typeCommand,2,CMD_READONLY|CMD_FAST,ACL_CATEGORY_KEYSPACE,{{CMD_KEY_RO,KSPEC_BS_INDEX,.bs.index={1},KSPEC_FK_RANGE,.fk.range={0,1,0}}},.args=TYPE_Args},
diff --git a/src/commands/acl-dryrun.json b/src/commands/acl-dryrun.json
new file mode 100644
index 000000000..544858c3a
--- /dev/null
+++ b/src/commands/acl-dryrun.json
@@ -0,0 +1,35 @@
+{
+    "DRYRUN": {
+        "summary": "Returns whether the user can execute the given command without executing the command.",
+        "complexity": "O(1).",
+        "group": "server",
+        "since": "7.0.0",
+        "arity": -4,
+        "container": "ACL",
+        "function": "aclCommand",
+        "history": [],
+        "command_flags": [
+            "ADMIN",
+            "NOSCRIPT",
+            "LOADING",
+            "STALE",
+            "SENTINEL"
+        ],
+        "arguments": [
+            {
+                "name": "username",
+                "type": "string"
+            },
+            {
+                "name": "command",
+                "type": "string"
+            },
+            {
+                "name": "arg",
+                "type": "string",
+                "optional": true,
+                "multiple": true
+            }
+        ]
+    }
+}
diff --git a/src/commands/acl-setuser.json b/src/commands/acl-setuser.json
index 789bc2d41..7f1f308df 100644
--- a/src/commands/acl-setuser.json
+++ b/src/commands/acl-setuser.json
@@ -11,6 +11,10 @@
             [
                 "6.2.0",
                 "Added Pub/Sub channel patterns."
+            ],
+            [
+                "7.0.0",
+                "Added selectors and key based permissions."
             ]
         ],
         "command_flags": [
diff --git a/src/commands/sort.json b/src/commands/sort.json
index 1c90ec7c2..26e380b42 100644
--- a/src/commands/sort.json
+++ b/src/commands/sort.json
@@ -21,8 +21,7 @@
             {
                 "flags": [
                     "RO",
-                    "ACCESS",
-                    "INCOMPLETE"
+                    "ACCESS"
                 ],
                 "begin_search": {
                     "index": {
@@ -39,6 +38,19 @@
             },
             {
                 "flags": [
+                    "RO",
+                    "ACCESS",
+                    "INCOMPLETE"
+                ],
+                "begin_search": {
+                    "unknown": null
+                },
+                "find_keys": {
+                    "unknown": null
+                }
+            },
+            {
+                "flags": [
                     "OW",
                     "UPDATE",
                     "INCOMPLETE"
diff --git a/src/commands/sort_ro.json b/src/commands/sort_ro.json
index ad78694f1..0858a126d 100644
--- a/src/commands/sort_ro.json
+++ b/src/commands/sort_ro.json
@@ -6,6 +6,7 @@
         "since": "7.0.0",
         "arity": -2,
         "function": "sortroCommand",
+        "get_keys_function": "sortROGetKeys",
         "command_flags": [
             "READONLY"
         ],
@@ -19,8 +20,7 @@
             {
                 "flags": [
                     "RO",
-                    "ACCESS",
-                    "INCOMPLETE"
+                    "ACCESS"
                 ],
                 "begin_search": {
                     "index": {
@@ -34,6 +34,19 @@
                         "limit": 0
                     }
                 }
+            },
+            {
+                "flags": [
+                    "RO",
+                    "ACCESS",
+                    "INCOMPLETE"
+                ],
+                "begin_search": {
+                    "unknown": null
+                },
+                "find_keys": {
+                    "unknown": null
+                }
             }
         ],
         "arguments": [
diff --git a/src/config.c b/src/config.c
index 46bd1ebb9..482a29617 100644
--- a/src/config.c
+++ b/src/config.c
@@ -122,7 +122,7 @@ configEnum oom_score_adj_enum[] = {
 };
 
 configEnum acl_pubsub_default_enum[] = {
-    {"allchannels", USER_FLAG_ALLCHANNELS},
+    {"allchannels", SELECTOR_FLAG_ALLCHANNELS},
     {"resetchannels", 0},
     {NULL, 0}
 };
@@ -2811,7 +2811,7 @@ standardConfig configs[] = {
     createEnumConfig("maxmemory-policy", NULL, MODIFIABLE_CONFIG, maxmemory_policy_enum, server.maxmemory_policy, MAXMEMORY_NO_EVICTION, NULL, NULL),
     createEnumConfig("appendfsync", NULL, MODIFIABLE_CONFIG, aof_fsync_enum, server.aof_fsync, AOF_FSYNC_EVERYSEC, NULL, NULL),
     createEnumConfig("oom-score-adj", NULL, MODIFIABLE_CONFIG, oom_score_adj_enum, server.oom_score_adj, OOM_SCORE_ADJ_NO, NULL, updateOOMScoreAdj),
-    createEnumConfig("acl-pubsub-default", NULL, MODIFIABLE_CONFIG, acl_pubsub_default_enum, server.acl_pubsub_default, USER_FLAG_ALLCHANNELS, NULL, NULL),
+    createEnumConfig("acl-pubsub-default", NULL, MODIFIABLE_CONFIG, acl_pubsub_default_enum, server.acl_pubsub_default, SELECTOR_FLAG_ALLCHANNELS, NULL, NULL),
     createEnumConfig("sanitize-dump-payload", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, sanitize_dump_payload_enum, server.sanitize_dump_payload, SANITIZE_DUMP_NO, NULL, NULL),
     createEnumConfig("enable-protected-configs", NULL, IMMUTABLE_CONFIG, protected_action_enum, server.enable_protected_configs, PROTECTED_ACTION_ALLOWED_NO, NULL, NULL),
     createEnumConfig("enable-debug-command", NULL, IMMUTABLE_CONFIG, protected_action_enum, server.enable_debug_cmd, PROTECTED_ACTION_ALLOWED_NO, NULL, NULL),
diff --git a/src/db.c b/src/db.c
index ff20f4687..94fbc0227 100644
--- a/src/db.c
+++ b/src/db.c
@@ -1653,7 +1653,7 @@ int expireIfNeeded(redisDb *db, robj *key, int force_delete_expired) {
  * This function must be called at least once before starting to populate
  * the result, and can be called repeatedly to enlarge the result array.
  */
-int *getKeysPrepareResult(getKeysResult *result, int numkeys) {
+keyReference *getKeysPrepareResult(getKeysResult *result, int numkeys) {
     /* GETKEYS_RESULT_INIT initializes keys to NULL, point it to the pre-allocated stack
      * buffer here. */
     if (!result->keys) {
@@ -1665,12 +1665,12 @@ int *getKeysPrepareResult(getKeysResult *result, int numkeys) {
     if (numkeys > result->size) {
         if (result->keys != result->keysbuf) {
             /* We're not using a static buffer, just (re)alloc */
-            result->keys = zrealloc(result->keys, numkeys * sizeof(int));
+            result->keys = zrealloc(result->keys, numkeys * sizeof(keyReference));
         } else {
             /* We are using a static buffer, copy its contents */
-            result->keys = zmalloc(numkeys * sizeof(int));
+            result->keys = zmalloc(numkeys * sizeof(keyReference));
             if (result->numkeys)
-                memcpy(result->keys, result->keysbuf, result->numkeys * sizeof(int));
+                memcpy(result->keys, result->keysbuf, result->numkeys * sizeof(keyReference));
         }
         result->size = numkeys;
     }
@@ -1678,12 +1678,183 @@ int *getKeysPrepareResult(getKeysResult *result, int numkeys) {
     return result->keys;
 }
 
+/* Returns a bitmask with all the flags found in any of the key specs of the command.
+ * The 'inv' argument means we'll return a mask with all flags that are missing in at least one spec. */
+int64_t getAllKeySpecsFlags(struct redisCommand *cmd, int inv) {
+    int64_t flags = 0;
+    for (int j = 0; j < cmd->key_specs_num; j++) {
+        keySpec *spec = cmd->key_specs + j;
+        flags |= inv? ~spec->flags : spec->flags;
+    }
+    return flags;
+}
+
+/* Fetch the keys based of the provided key specs. Returns the number of keys found, or -1 on error.
+ * There are several flags that can be used to modify how this function finds keys in a command.
+ * 
+ * GET_KEYSPEC_INCLUDE_CHANNELS: Return channels as if they were keys.
+ * GET_KEYSPEC_RETURN_PARTIAL:   Skips invalid and incomplete keyspecs but returns the keys
+ *                               found in other valid keyspecs. 
+ */
+int getKeysUsingKeySpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) {
+    int j, i, k = 0, last, first, step;
+    keyReference *keys;
+
+    for (j = 0; j < cmd->key_specs_num; j++) {
+        keySpec *spec = cmd->key_specs + j;
+        serverAssert(spec->begin_search_type != KSPEC_BS_INVALID);
+        /* Skip specs that represent channels instead of keys */
+        if (spec->flags & (CMD_KEY_CHANNEL) && !(search_flags & GET_KEYSPEC_INCLUDE_CHANNELS)) {
+            continue;
+        }
+
+        first = 0;
+        if (spec->begin_search_type == KSPEC_BS_INDEX) {
+            first = spec->bs.index.pos;
+        } else if (spec->begin_search_type == KSPEC_BS_KEYWORD) {
+            int start_index = spec->bs.keyword.startfrom > 0 ? spec->bs.keyword.startfrom : argc+spec->bs.keyword.startfrom;
+            int end_index = spec->bs.keyword.startfrom > 0 ? argc-1: 1;
+            for (i = start_index; i != end_index; i = start_index <= end_index ? i + 1 : i - 1) {
+                if (i >= argc || i < 1)
+                    break;
+                if (!strcasecmp((char*)argv[i]->ptr,spec->bs.keyword.keyword)) {
+                    first = i+1;
+                    break;
+                }
+            }
+            /* keyword not found */
+            if (!first) {
+                continue;
+            }
+        } else {
+            /* unknown spec */
+            goto invalid_spec;
+        }
+
+        if (spec->find_keys_type == KSPEC_FK_RANGE) {
+            step = spec->fk.range.keystep;
+            if (spec->fk.range.lastkey >= 0) {
+                last = first + spec->fk.range.lastkey;
+            } else {
+                if (!spec->fk.range.limit) {
+                    last = argc + spec->fk.range.lastkey;
+                } else {
+                    serverAssert(spec->fk.range.lastkey == -1);
+                    last = first + ((argc-first)/spec->fk.range.limit + spec->fk.range.lastkey);
+                }
+            }
+        } else if (spec->find_keys_type == KSPEC_FK_KEYNUM) {
+            step = spec->fk.keynum.keystep;
+            long long numkeys;
+            if (spec->fk.keynum.keynumidx >= argc)
+                goto invalid_spec;
+
+            sds keynum_str = argv[first + spec->fk.keynum.keynumidx]->ptr;
+            if (!string2ll(keynum_str,sdslen(keynum_str),&numkeys) || numkeys < 0) {
+                /* Unable to parse the numkeys argument or it was invalid */
+                goto invalid_spec;
+            }
+
+            first += spec->fk.keynum.firstkey;
+            last = first + (int)numkeys-1;
+        } else {
+            /* unknown spec */
+            goto invalid_spec;
+        }
+
+        int count = ((last - first)+1);
+        keys = getKeysPrepareResult(result, count);
+
+        /* First or last is out of bounds, which indicates a syntax error */
+        if (last >= argc || last < first || first >= argc) {
+            goto invalid_spec;
+        }
+
+        for (i = first; i <= last; i += step) {
+            if (i >= argc || i < first) {
+                /* Modules commands, and standard commands with a not fixed number
+                 * of arguments (negative arity parameter) do not have dispatch
+                 * time arity checks, so we need to handle the case where the user
+                 * passed an invalid number of arguments here. In this case we
+                 * return no keys and expect the command implementation to report
+                 * an arity or syntax error. */
+                if (cmd->flags & CMD_MODULE || cmd->arity < 0) {
+                    continue;
+                } else {
+                    serverPanic("Redis built-in command declared keys positions not matching the arity requirements.");
+                }
+            }
+            keys[k].pos = i;
+            keys[k++].flags = spec->flags;
+        }
+
+        /* Done with this spec */
+        continue;
+
+invalid_spec:
+        if (search_flags & GET_KEYSPEC_RETURN_PARTIAL) {
+            continue;
+        } else {
+            result->numkeys = 0;
+            return -1;
+        }
+    }
+
+    result->numkeys = k;
+    return k;
+}
+
+/* Return all the arguments that are keys in the command passed via argc / argv. 
+ * This function will eventually replace getKeysFromCommand.
+ *
+ * The command returns the positions of all the key arguments inside the array,
+ * so the actual return value is a heap allocated array of integers. The
+ * length of the array is returned by reference into *numkeys.
+ * 
+ * Along with the position, this command also returns the flags that are
+ * associated with how Redis will access the key.
+ *
+ * 'cmd' must be point to the corresponding entry into the redisCommand
+ * table, according to the command name in argv[0].
+ *
+ * This function uses the command's key specs, which contain the key-spec flags,
+ * (e.g. RO / RW) and only resorts to the command-specific helper function if
+ * any of the keys-specs are marked as INCOMPLETE. */
+int getKeysFromCommandWithSpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) {
+    if (cmd->flags & CMD_MODULE_GETKEYS) {
+        return moduleGetCommandKeysViaAPI(cmd,argv,argc,result);
+    } else {
+        if (!(getAllKeySpecsFlags(cmd, 0) & CMD_KEY_INCOMPLETE)) {
+            int ret = getKeysUsingKeySpecs(cmd,argv,argc,search_flags,result);
+            if (ret >= 0)
+                return ret;
+        }
+        if (!(cmd->flags & CMD_MODULE) && cmd->getkeys_proc)
+            return cmd->getkeys_proc(cmd,argv,argc,result);
+        return 0;
+    }
+}
+
+/* This function returns a sanity check if the command may have keys. */
+int doesCommandHaveKeys(struct redisCommand *cmd) {
+    return (!(cmd->flags & CMD_MODULE) && cmd->getkeys_proc) || /* has getkeys_proc (non modules) */
+        (cmd->flags & CMD_MODULE_GETKEYS) ||                    /* module with GETKEYS */
+        (getAllKeySpecsFlags(cmd, 1) & CMD_KEY_CHANNEL);       /* has at least one key-spec not marked as CHANNEL */
+}
+
 /* The base case is to use the keys position as given in the command table
  * (firstkey, lastkey, step).
  * This function works only on command with the legacy_range_key_spec,
- * all other commands should be handled by getkeys_proc. */
+ * all other commands should be handled by getkeys_proc. 
+ * 
+ * If the commands keyspec is incomplete, no keys will be returned, and the provided
+ * keys function should be called instead.
+ * 
+ * NOTE: This function does not guarantee populating the flags for 
+ * the keys, in order to get flags you should use getKeysUsingKeySpecs. */
 int getKeysUsingLegacyRangeSpec(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
-    int j, i = 0, last, first, step, *keys;
+    int j, i = 0, last, first, step;
+    keyReference *keys;
     UNUSED(argv);
 
     if (cmd->legacy_range_key_spec.begin_search_type == KSPEC_BS_INVALID) {
@@ -1703,7 +1874,7 @@ int getKeysUsingLegacyRangeSpec(struct redisCommand *cmd, robj **argv, int argc,
     keys = getKeysPrepareResult(result, count);
 
     for (j = first; j <= last; j += step) {
-        if (j >= argc) {
+        if (j >= argc || j < first) {
             /* Modules commands, and standard commands with a not fixed number
              * of arguments (negative arity parameter) do not have dispatch
              * time arity checks, so we need to handle the case where the user
@@ -1717,7 +1888,9 @@ int getKeysUsingLegacyRangeSpec(struct redisCommand *cmd, robj **argv, int argc,
                 serverPanic("Redis built-in command declared keys positions not matching the arity requirements.");
             }
         }
-        keys[i++] = j;
+        keys[i].pos = j;
+        /* Flags are omitted from legacy key specs */
+        keys[i++].flags = 0;
     }
     result->numkeys = i;
     return i;
@@ -1761,10 +1934,12 @@ void getKeysFreeResult(getKeysResult *result) {
  * 'keyCountOfs': num-keys index.
  * 'firstKeyOfs': firstkey index.
  * 'keyStep': the interval of each key, usually this value is 1.
- * */
+ * 
+ * The commands using this functoin have a fully defined keyspec, so returning flags isn't needed. */
 int genericGetKeys(int storeKeyOfs, int keyCountOfs, int firstKeyOfs, int keyStep,
                     robj **argv, int argc, getKeysResult *result) {
-    int i, num, *keys;
+    int i, num;
+    keyReference *keys;
 
     num = atoi(argv[keyCountOfs]->ptr);
     /* Sanity check. Don't return any key if the command is going to
@@ -1779,9 +1954,15 @@ int genericGetKeys(int storeKeyOfs, int keyCountOfs, int firstKeyOfs, int keySte
     result->numkeys = numkeys;
 
     /* Add all key positions for argv[firstKeyOfs...n] to keys[] */
-    for (i = 0; i < num; i++) keys[i] = firstKeyOfs+(i*keyStep);
-
-    if (storeKeyOfs) keys[num] = storeKeyOfs;
+    for (i = 0; i < num; i++) {
+        keys[i].pos = firstKeyOfs+(i*keyStep);
+        keys[i].flags = 0;
+    } 
+
+    if (storeKeyOfs) {
+        keys[num].pos = storeKeyOfs;
+        keys[num].flags = 0;
+    } 
     return result->numkeys;
 }
 
@@ -1830,20 +2011,46 @@ int bzmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult
     return genericGetKeys(0, 2, 3, 1, argv, argc, result);
 }
 
+/* Helper function to extract keys from the SORT RO command.
+ *
+ * SORT <sort-key>
+ *
+ * The second argument of SORT is always a key, however an arbitrary number of
+ * keys may be accessed while doing the sort (the BY and GET args), so the
+ * key-spec declares incomplete keys which is why we have to provide a concrete
+ * implementation to fetch the keys.
+ *
+ * This command declares incomplete keys, so the flags are correctly set for this function */
+int sortROGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    keyReference *keys;
+    UNUSED(cmd);
+    UNUSED(argv);
+    UNUSED(argc);
+
+    keys = getKeysPrepareResult(result, 1);
+    keys[0].pos = 1; /* <sort-key> is always present. */
+    keys[0].flags = CMD_KEY_RO | CMD_KEY_ACCESS;
+    return 1;
+}
+
 /* Helper function to extract keys from the SORT command.
  *
  * SORT <sort-key> ... STORE <store-key> ...
  *
  * The first argument of SORT is always a key, however a list of options
  * follow in SQL-alike style. Here we parse just the minimum in order to
- * correctly identify keys in the "STORE" option. */
+ * correctly identify keys in the "STORE" option. 
+ * 
+ * This command declares incomplete keys, so the flags are correctly set for this function */
 int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
-    int i, j, num, *keys, found_store = 0;
+    int i, j, num, found_store = 0;
+    keyReference *keys;
     UNUSED(cmd);
 
     num = 0;
     keys = getKeysPrepareResult(result, 2); /* Alloc 2 places for the worst case. */
-    keys[num++] = 1; /* <sort-key> is always present. */
+    keys[num].pos = 1; /* <sort-key> is always present. */
+    keys[num++].flags = CMD_KEY_RO | CMD_KEY_ACCESS;
 
     /* Search for STORE option. By default we consider options to don't
      * have arguments, so if we find an unknown option name we scan the
@@ -1869,7 +2076,8 @@ int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *
                  * to be sure to process the *last* "STORE" option if multiple
                  * ones are provided. This is same behavior as SORT. */
                 found_store = 1;
-                keys[num] = i+1; /* <store-key> */
+                keys[num].pos = i+1; /* <store-key> */
+                keys[num].flags = CMD_KEY_OW | CMD_KEY_UPDATE;
                 break;
             }
         }
@@ -1878,8 +2086,10 @@ int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *
     return result->numkeys;
 }
 
+/* This command declares incomplete keys, so the flags are correctly set for this function */
 int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
-    int i, num, first, *keys;
+    int i, num, first;
+    keyReference *keys;
     UNUSED(cmd);
 
     /* Assume the obvious form. */
@@ -1900,7 +2110,10 @@ int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResul
     }
 
     keys = getKeysPrepareResult(result, num);
-    for (i = 0; i < num; i++) keys[i] = first+i;
+    for (i = 0; i < num; i++) {
+        keys[i].pos = first+i;
+        keys[i].flags = CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_DELETE;
+    } 
     result->numkeys = num;
     return num;
 }
@@ -1908,9 +2121,12 @@ int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResul
 /* Helper function to extract keys from following commands:
  * GEORADIUS key x y radius unit [WITHDIST] [WITHHASH] [WITHCOORD] [ASC|DESC]
  *                             [COUNT count] [STORE key] [STOREDIST key]
- * GEORADIUSBYMEMBER key member radius unit ... options ... */
+ * GEORADIUSBYMEMBER key member radius unit ... options ...
+ * 
+ * This command has a fully defined keyspec, so returning flags isn't needed. */
 int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
-    int i, num, *keys;
+    int i, num;
+    keyReference *keys;
     UNUSED(cmd);
 
     /* Check for the presence of the stored key in the command */
@@ -1935,18 +2151,23 @@ int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysRes
     keys = getKeysPrepareResult(result, num);
 
     /* Add all key positions to keys[] */
-    keys[0] = 1;
+    keys[0].pos = 1;
+    keys[0].flags = 0;
     if(num > 1) {
-         keys[1] = stored_key;
+         keys[1].pos = stored_key;
+         keys[1].flags = 0;
     }
     result->numkeys = num;
     return num;
 }
 
 /* XREAD [BLOCK <milliseconds>] [COUNT <count>] [GROUP <groupname> <ttl>]
- *       STREAMS key_1 key_2 ... key_N ID_1 ID_2 ... ID_N */
+ *       STREAMS key_1 key_2 ... key_N ID_1 ID_2 ... ID_N
+ *
+ * This command has a fully defined keyspec, so returning flags isn't needed. */
 int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
-    int i, num = 0, *keys;
+    int i, num = 0;
+    keyReference *keys;
     UNUSED(cmd);
 
     /* We need to parse the options of the command in order to seek the first
@@ -1982,7 +2203,10 @@ int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult
                  there are also the IDs, one per key. */
 
     keys = getKeysPrepareResult(result, num);
-    for (i = streams_pos+1; i < argc-num; i++) keys[i-streams_pos-1] = i;
+    for (i = streams_pos+1; i < argc-num; i++) {
+        keys[i-streams_pos-1].pos = i;
+        keys[i-streams_pos-1].flags = 0; 
+    } 
     result->numkeys = num;
     return num;
 }
diff --git a/src/module.c b/src/module.c
index 5626310a3..2e2652717 100644
--- a/src/module.c
+++ b/src/module.c
@@ -811,7 +811,7 @@ void RM_KeyAtPos(RedisModuleCtx *ctx, int pos) {
         getKeysPrepareResult(res, newsize);
     }
 
-    res->keys[res->numkeys++] = pos;
+    res->keys[res->numkeys++].pos = pos;
 }
 
 /* Helper for RM_CreateCommand(). Turns a string representing command
@@ -864,10 +864,10 @@ int64_t commandKeySpecsFlagsFromString(const char *s) {
         else if (!strcasecmp(t,"RW")) flags |= CMD_KEY_RW;
         else if (!strcasecmp(t,"OW")) flags |= CMD_KEY_OW;
         else if (!strcasecmp(t,"RM")) flags |= CMD_KEY_RM;
-        else if (!strcasecmp(t,"ACCESS")) flags |= CMD_KEY_ACCESS;
-        else if (!strcasecmp(t,"INSERT")) flags |= CMD_KEY_INSERT;
-        else if (!strcasecmp(t,"UPDATE")) flags |= CMD_KEY_UPDATE;
-        else if (!strcasecmp(t,"DELETE")) flags |= CMD_KEY_DELETE;
+        else if (!strcasecmp(t,"access")) flags |= CMD_KEY_ACCESS;
+        else if (!strcasecmp(t,"insert")) flags |= CMD_KEY_INSERT;
+        else if (!strcasecmp(t,"update")) flags |= CMD_KEY_UPDATE;
+        else if (!strcasecmp(t,"delete")) flags |= CMD_KEY_DELETE;
         else if (!strcasecmp(t,"channel")) flags |= CMD_KEY_CHANNEL;
         else if (!strcasecmp(t,"incomplete")) flags |= CMD_KEY_INCOMPLETE;
         else break;
@@ -1218,14 +1218,14 @@ int moduleSetCommandKeySpecFindKeys(RedisModuleCommand *command, int index, keyS
  *      if (RedisModule_CreateCommand(ctx,"kspec.smove",kspec_legacy,"",0,0,0) == REDISMODULE_ERR)
  *          return REDISMODULE_ERR;
  *
- *      if (RedisModule_AddCommandKeySpec(ctx,"kspec.smove","read write",&spec_id) == REDISMODULE_ERR)
+ *      if (RedisModule_AddCommandKeySpec(ctx,"kspec.smove","RW access delete",&spec_id) == REDISMODULE_ERR)
  *          return REDISMODULE_ERR;
  *      if (RedisModule_SetCommandKeySpecBeginSearchIndex(ctx,"kspec.smove",spec_id,1) == REDISMODULE_ERR)
  *          return REDISMODULE_ERR;
  *      if (RedisModule_SetCommandKeySpecFindKeysRange(ctx,"kspec.smove",spec_id,0,1,0) == REDISMODULE_ERR)
  *          return REDISMODULE_ERR;
  *
- *      if (RedisModule_AddCommandKeySpec(ctx,"kspec.smove","write",&spec_id) == REDISMODULE_ERR)
+ *      if (RedisModule_AddCommandKeySpec(ctx,"kspec.smove","RW insert",&spec_id) == REDISMODULE_ERR)
  *          return REDISMODULE_ERR;
  *      if (RedisModule_SetCommandKeySpecBeginSearchIndex(ctx,"kspec.smove",spec_id,2) == REDISMODULE_ERR)
  *          return REDISMODULE_ERR;
@@ -1237,7 +1237,7 @@ int moduleSetCommandKeySpecFindKeys(RedisModuleCommand *command, int index, keyS
  *
  * Example:
  *
- *      RedisModule_AddCommandKeySpec(ctx,"module.config|get","read",&spec_id)
+ *      RedisModule_AddCommandKeySpec(ctx,"module.object|encoding","RO",&spec_id)
  *
  * Returns REDISMODULE_OK on success
  */
@@ -7793,13 +7793,31 @@ int RM_ACLCheckCommandPermissions(RedisModuleUser *user, RedisModuleString **arg
     return REDISMODULE_OK;
 }
 
-/* Check if the key can be accessed by the user, according to the ACLs associated with it.
+/* Check if the key can be accessed by the user, according to the ACLs associated with it
+ * and the flags used. The supported flags are:
  *
- * If the user can access the key, REDISMODULE_OK is returned, otherwise
- * REDISMODULE_ERR is returned. */
-int RM_ACLCheckKeyPermissions(RedisModuleUser *user, RedisModuleString *key) {
-    if (ACLCheckKey(user->user, key->ptr, sdslen(key->ptr)) != ACL_OK)
+ * REDISMODULE_KEY_PERMISSION_READ: Can the module read data from the key.
+ * REDISMODULE_KEY_PERMISSION_WRITE: Can the module write data to the key.
+ *
+ * On success a REDISMODULE_OK is returned, otherwise
+ * REDISMODULE_ERR is returned and errno is set to the following values:
+ * 
+ * * EINVAL: The provided flags are invalid.
+ * * EACCESS: The user does not have permission to access the key.
+ */
+int RM_ACLCheckKeyPermissions(RedisModuleUser *user, RedisModuleString *key, int flags) {
+    int acl_flags = 0;
+    if (flags & REDISMODULE_KEY_PERMISSION_READ) acl_flags |= ACL_READ_PERMISSION;
+    if (flags & REDISMODULE_KEY_PERMISSION_WRITE) acl_flags |= ACL_WRITE_PERMISSION;
+    if (!acl_flags || ((flags & REDISMODULE_KEY_PERMISSION_ALL) != flags)) {
+        errno = EINVAL;
         return REDISMODULE_ERR;
+    }
+
+    if (ACLUserCheckKeyPerm(user->user, key->ptr, sdslen(key->ptr), acl_flags) != ACL_OK) {
+        errno = EACCES;
+        return REDISMODULE_ERR;
+    }
 
     return REDISMODULE_OK;
 }
@@ -7811,7 +7829,7 @@ int RM_ACLCheckKeyPermissions(RedisModuleUser *user, RedisModuleString *key) {
  * If the user can access the pubsub channel, REDISMODULE_OK is returned, otherwise
  * REDISMODULE_ERR is returned. */
 int RM_ACLCheckChannelPermissions(RedisModuleUser *user, RedisModuleString *ch, int literal) {
-    if (ACLCheckPubsubChannelPerm(ch->ptr, user->user->channels, literal) != ACL_OK)
+    if (ACLUserCheckChannelPerm(user->user, ch->ptr, literal) != ACL_OK)
         return REDISMODULE_ERR;
 
     return REDISMODULE_OK;
@@ -10490,17 +10508,11 @@ int *RM_GetCommandKeys(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
         return NULL;
     }
 
-    if (result.keys == result.keysbuf) {
-        /* If the result is using a stack based array, copy it. */
-        unsigned long int size = sizeof(int) * result.numkeys;
-        res = zmalloc(size);
-        memcpy(res, result.keys, size);
-    } else {
-        /* We return the heap based array and intentionally avoid calling
-         * getKeysFreeResult() here, as it is the caller's responsibility
-         * to free this array.
-         */
-        res = result.keys;
+    /* The return value here expects an array of key positions */
+    unsigned long int size = sizeof(int) * result.numkeys;
+    res = zmalloc(size);
+    for (int i = 0; i < result.numkeys; i++) {
+        res[i] = result.keys[i].pos;
     }
 
     return res;
diff --git a/src/redismodule.h b/src/redismodule.h
index 15990a04f..4bcad5111 100644
--- a/src/redismodule.h
+++ b/src/redismodule.h
@@ -261,6 +261,12 @@ typedef enum {
 #define REDISMODULE_CMD_ARG_MULTIPLE        (1<<1) /* The argument may repeat itself (like key in DEL) */
 #define REDISMODULE_CMD_ARG_MULTIPLE_TOKEN  (1<<2) /* The argument may repeat itself, and so does its token (like `GET pattern` in SORT) */
 
+/* Redis ACL key permission flags, which specify which permissions a module
+ * needs on a key. */
+#define REDISMODULE_KEY_PERMISSION_READ (1<<0) 
+#define REDISMODULE_KEY_PERMISSION_WRITE (1<<1)
+#define REDISMODULE_KEY_PERMISSION_ALL (REDISMODULE_KEY_PERMISSION_READ | REDISMODULE_KEY_PERMISSION_WRITE)
+
 /* Eventloop definitions. */
 #define REDISMODULE_EVENTLOOP_READABLE 1
 #define REDISMODULE_EVENTLOOP_WRITABLE 2
@@ -978,7 +984,7 @@ REDISMODULE_API int (*RedisModule_SetModuleUserACL)(RedisModuleUser *user, const
 REDISMODULE_API RedisModuleString * (*RedisModule_GetCurrentUserName)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
 REDISMODULE_API RedisModuleUser * (*RedisModule_GetModuleUserFromUserName)(RedisModuleString *name) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_ACLCheckCommandPermissions)(RedisModuleUser *user, RedisModuleString **argv, int argc) REDISMODULE_ATTR;
-REDISMODULE_API int (*RedisModule_ACLCheckKeyPermissions)(RedisModuleUser *user, RedisModuleString *key) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_ACLCheckKeyPermissions)(RedisModuleUser *user, RedisModuleString *key, int flags) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_ACLCheckChannelPermissions)(RedisModuleUser *user, RedisModuleString *ch, int literal) REDISMODULE_ATTR;
 REDISMODULE_API void (*RedisModule_ACLAddLogEntry)(RedisModuleCtx *ctx, RedisModuleUser *user, RedisModuleString *object) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_AuthenticateClientWithACLUser)(RedisModuleCtx *ctx, const char *name, size_t len, RedisModuleUserChangedFunc callback, void *privdata, uint64_t *client_id) REDISMODULE_ATTR;
diff --git a/src/server.c b/src/server.c
index 4561c99d9..5de6d61c5 100644
--- a/src/server.c
+++ b/src/server.c
@@ -4504,7 +4504,7 @@ void getKeysSubcommand(client *c) {
         }
     } else {
         addReplyArrayLen(c,result.numkeys);
-        for (j = 0; j < result.numkeys; j++) addReplyBulk(c,c->argv[result.keys[j]+2]);
+        for (j = 0; j < result.numkeys; j++) addReplyBulk(c,c->argv[result.keys[j].pos+2]);
     }
     getKeysFreeResult(&result);
 }
diff --git a/src/server.h b/src/server.h
index cb884e370..084abdecd 100644
--- a/src/server.h
+++ b/src/server.h
@@ -995,54 +995,32 @@ typedef struct readyList {
                                            is USER_COMMAND_BITS_COUNT-1. */
 #define USER_FLAG_ENABLED (1<<0)        /* The user is active. */
 #define USER_FLAG_DISABLED (1<<1)       /* The user is disabled. */
-#define USER_FLAG_ALLKEYS (1<<2)        /* The user can mention any key. */
-#define USER_FLAG_ALLCOMMANDS (1<<3)    /* The user can run all commands. */
-#define USER_FLAG_NOPASS      (1<<4)    /* The user requires no password, any
+#define USER_FLAG_NOPASS (1<<2)         /* The user requires no password, any
                                            provided password will work. For the
                                            default user, this also means that
                                            no AUTH is needed, and every
                                            connection is immediately
                                            authenticated. */
-#define USER_FLAG_ALLCHANNELS (1<<5)    /* The user can mention any Pub/Sub
-                                           channel. */
-#define USER_FLAG_SANITIZE_PAYLOAD (1<<6)       /* The user require a deep RESTORE
+#define USER_FLAG_SANITIZE_PAYLOAD (1<<3)       /* The user require a deep RESTORE
                                                  * payload sanitization. */
-#define USER_FLAG_SANITIZE_PAYLOAD_SKIP (1<<7)  /* The user should skip the
+#define USER_FLAG_SANITIZE_PAYLOAD_SKIP (1<<4)  /* The user should skip the
                                                  * deep sanitization of RESTORE
                                                  * payload. */
 
+#define SELECTOR_FLAG_ROOT (1<<0)           /* This is the root user permission
+                                             * selector. */
+#define SELECTOR_FLAG_ALLKEYS (1<<1)        /* The user can mention any key. */
+#define SELECTOR_FLAG_ALLCOMMANDS (1<<2)    /* The user can run all commands. */
+#define SELECTOR_FLAG_ALLCHANNELS (1<<3)    /* The user can mention any Pub/Sub
+                                               channel. */
+
 typedef struct {
     sds name;       /* The username as an SDS string. */
-    uint64_t flags; /* See USER_FLAG_* */
-
-    /* The bit in allowed_commands is set if this user has the right to
-     * execute this command.
-     *
-     * If the bit for a given command is NOT set and the command has
-     * allowed first-args, Redis will also check allowed_firstargs in order to
-     * understand if the command can be executed. */
-    uint64_t allowed_commands[USER_COMMAND_BITS_COUNT/64];
-
-    /* allowed_firstargs is used by ACL rules to block access to a command unless a
-     * specific argv[1] is given (or argv[2] in case it is applied on a sub-command).
-     * For example, a user can use the rule "-select +select|0" to block all
-     * SELECT commands, except "SELECT 0".
-     * And for a sub-command: "+config -config|set +config|set|loglevel"
-     *
-     * For each command ID (corresponding to the command bit set in allowed_commands),
-     * This array points to an array of SDS strings, terminated by a NULL pointer,
-     * with all the first-args that are allowed for this command. When no first-arg
-     * matching is used, the field is just set to NULL to avoid allocating
-     * USER_COMMAND_BITS_COUNT pointers. */
-    sds **allowed_firstargs;
+    uint32_t flags; /* See USER_FLAG_* */
     list *passwords; /* A list of SDS valid passwords for this user. */
-    list *patterns;  /* A list of allowed key patterns. If this field is NULL
-                        the user cannot mention any key in a command, unless
-                        the flag ALLKEYS is set in the user. */
-    list *channels;  /* A list of allowed Pub/Sub channel patterns. If this
-                        field is NULL the user cannot mention any channel in a
-                        `PUBLISH` or [P][UNSUBSCRIBE] command, unless the flag
-                        ALLCHANNELS is set in the user. */
+    list *selectors; /* A list of selectors this user validates commands
+                        against. This list will always contain at least
+                        one selector for backwards compatibility. */
 } user;
 
 /* With multiplexing we need to take per-client state.
@@ -1913,16 +1891,22 @@ struct redisServer {
 
 #define MAX_KEYS_BUFFER 256
 
+typedef struct {
+    int pos; /* The position of the key within the client array */
+    int flags; /* The flags associted with the key access, see
+                  CMD_KEY_* for more information */
+} keyReference;
+
 /* A result structure for the various getkeys function calls. It lists the
  * keys as indices to the provided argv.
  */
 typedef struct {
-    int keysbuf[MAX_KEYS_BUFFER];       /* Pre-allocated buffer, to save heap allocations */
-    int *keys;                          /* Key indices array, points to keysbuf or heap */
+    keyReference keysbuf[MAX_KEYS_BUFFER];       /* Pre-allocated buffer, to save heap allocations */
+    keyReference *keys;                          /* Key indices array, points to keysbuf or heap */
     int numkeys;                        /* Number of key indices return */
     int size;                           /* Available array size */
 } getKeysResult;
-#define GETKEYS_RESULT_INIT { {0}, NULL, 0, MAX_KEYS_BUFFER }
+#define GETKEYS_RESULT_INIT { {{0}}, NULL, 0, MAX_KEYS_BUFFER }
 
 /* Key specs definitions.
  *
@@ -2714,14 +2698,19 @@ void ACLInit(void);
 #define ACL_LOG_CTX_MULTI 2
 #define ACL_LOG_CTX_MODULE 3
 
+/* ACL key permission types */
+#define ACL_READ_PERMISSION (1<<0)
+#define ACL_WRITE_PERMISSION (1<<1)
+#define ACL_ALL_PERMISSION (ACL_READ_PERMISSION|ACL_WRITE_PERMISSION)
+
 int ACLCheckUserCredentials(robj *username, robj *password);
 int ACLAuthenticateUser(client *c, robj *username, robj *password);
 unsigned long ACLGetCommandID(const char *cmdname);
 void ACLClearCommandID(void);
 user *ACLGetUserByName(const char *name, size_t namelen);
-int ACLCheckKey(const user *u, const char *key, int keylen);
-int ACLCheckPubsubChannelPerm(sds channel, list *allowed, int literal);
-int ACLCheckAllUserCommandPerm(const user *u, struct redisCommand *cmd, robj **argv, int argc, int *idxptr);
+int ACLUserCheckKeyPerm(user *u, const char *key, int keylen, int flags);
+int ACLUserCheckChannelPerm(user *u, sds channel, int literal);
+int ACLCheckAllUserCommandPerm(user *u, struct redisCommand *cmd, robj **argv, int argc, int *idxptr);
 int ACLCheckAllPerm(client *c, int *idxptr);
 int ACLSetUser(user *u, const char *op, ssize_t oplen);
 uint64_t ACLGetCommandCategoryFlagByName(const char *name);
@@ -3003,9 +2992,14 @@ void freeObjAsync(robj *key, robj *obj, int dbid);
 void freeReplicationBacklogRefMemAsync(list *blocks, rax *index);
 
 /* API to get key arguments from commands */
-int *getKeysPrepareResult(getKeysResult *result, int numkeys);
+#define GET_KEYSPEC_DEFAULT 0
+#define GET_KEYSPEC_INCLUDE_CHANNELS (1<<0) /* Consider channels as keys */
+#define GET_KEYSPEC_RETURN_PARTIAL (1<<1) /* Return all keys that can be found */
+
+int getKeysFromCommandWithSpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result);
+keyReference *getKeysPrepareResult(getKeysResult *result, int numkeys);
 int getKeysFromCommand(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
-int getChannelsFromCommand(struct redisCommand *cmd, int argc, getKeysResult *result);
+int doesCommandHaveKeys(struct redisCommand *cmd);
 void getKeysFreeResult(getKeysResult *result);
 int sintercardGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysResult *result);
 int zunionInterDiffGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysResult *result);
@@ -3013,6 +3007,7 @@ int zunionInterDiffStoreGetKeys(struct redisCommand *cmd,robj **argv, int argc,
 int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int functionGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
+int sortROGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
diff --git a/src/tracking.c b/src/tracking.c
index bb36f742d..b86d984a7 100644
--- a/src/tracking.c
+++ b/src/tracking.c
@@ -235,10 +235,10 @@ void trackingRememberKeys(client *c) {
         return;
     }
 
-    int *keys = result.keys;
+    keyReference *keys = result.keys;
 
     for(int j = 0; j < numkeys; j++) {
-        int idx = keys[j];
+        int idx = keys[j].pos;
         sds sdskey = c->argv[idx]->ptr;
         rax *ids = raxFind(TrackingTable,(unsigned char*)sdskey,sdslen(sdskey));
         if (ids == raxNotFound) {
diff --git a/tests/assets/userwithselectors.acl b/tests/assets/userwithselectors.acl
new file mode 100644
index 000000000..5d42957d3
--- /dev/null
+++ b/tests/assets/userwithselectors.acl
@@ -0,0 +1,2 @@
+user alice on (+get ~rw*)
+user bob on (+set %W~w*) (+get %R~r*)
\ No newline at end of file
diff --git a/tests/modules/aclcheck.c b/tests/modules/aclcheck.c
index fafb645a7..cc8d263fd 100644
--- a/tests/modules/aclcheck.c
+++ b/tests/modules/aclcheck.c
@@ -2,17 +2,33 @@
 #include "redismodule.h"
 #include <errno.h>
 #include <assert.h>
+#include <string.h>
+#include <strings.h>
 
 /* A wrap for SET command with ACL check on the key. */
 int set_aclcheck_key(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
-    if (argc < 3) {
+    if (argc < 4) {
         return RedisModule_WrongArity(ctx);
     }
 
+    int permissions;
+    const char *flags = RedisModule_StringPtrLen(argv[1], NULL);
+
+    if (!strcasecmp(flags, "W")) {
+        permissions = REDISMODULE_KEY_PERMISSION_WRITE;
+    } else if (!strcasecmp(flags, "R")) {
+        permissions = REDISMODULE_KEY_PERMISSION_READ;
+    } else if (!strcasecmp(flags, "*")) {
+        permissions = REDISMODULE_KEY_PERMISSION_ALL;
+    } else {
+        RedisModule_ReplyWithError(ctx, "INVALID FLAGS");
+        return REDISMODULE_OK;
+    }
+
     /* Check that the key can be accessed */
     RedisModuleString *user_name = RedisModule_GetCurrentUserName(ctx);
     RedisModuleUser *user = RedisModule_GetModuleUserFromUserName(user_name);
-    int ret = RedisModule_ACLCheckKeyPermissions(user, argv[1]);
+    int ret = RedisModule_ACLCheckKeyPermissions(user, argv[2], permissions);
     if (ret != 0) {
         RedisModule_ReplyWithError(ctx, "DENIED KEY");
         RedisModule_FreeModuleUser(user);
@@ -20,7 +36,7 @@ int set_aclcheck_key(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
         return REDISMODULE_OK;
     }
 
-    RedisModuleCallReply *rep = RedisModule_Call(ctx, "SET", "v", argv + 1, argc - 1);
+    RedisModuleCallReply *rep = RedisModule_Call(ctx, "SET", "v", argv + 2, argc - 2);
     if (!rep) {
         RedisModule_ReplyWithError(ctx, "NULL reply returned");
     } else {
diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl
index dd6af6342..1a5096937 100644
--- a/tests/test_helper.tcl
+++ b/tests/test_helper.tcl
@@ -37,6 +37,7 @@ set ::all_tests {
     unit/quit
     unit/aofrw
     unit/acl
+    unit/acl-v2
     unit/latency-monitor
     integration/block-repl
     integration/replication
diff --git a/tests/unit/acl-v2.tcl b/tests/unit/acl-v2.tcl
new file mode 100644
index 000000000..d0841b474
--- /dev/null
+++ b/tests/unit/acl-v2.tcl
@@ -0,0 +1,298 @@
+start_server {tags {"acl external:skip"}} {
+    set r2 [redis_client]
+    test {Test basic multiple selectors} {
+        r ACL SETUSER selector-1 on -@all resetkeys nopass
+        $r2 auth selector-1 password
+        catch {$r2 ping} err
+        assert_match "*NOPERM*command*" $err
+        catch {$r2 set write::foo bar} err
+        assert_match "*NOPERM*command*" $err
+        catch {$r2 get read::foo} err
+        assert_match "*NOPERM*command*" $err
+
+        r ACL SETUSER selector-1 (+@write ~write::*) (+@read ~read::*)
+        catch {$r2 ping} err
+        assert_equal "OK" [$r2 set write::foo bar]
+        assert_equal "" [$r2 get read::foo]
+        catch {$r2 get write::foo} err
+        assert_match "*NOPERM*keys*" $err
+        catch {$r2 set read::foo bar} err
+        assert_match "*NOPERM*keys*" $err
+    }
+
+    test {Test ACL selectors by default have no permissions (except channels)} {
+        r ACL SETUSER selector-default reset ()
+        set user [r ACL GETUSER "selector-default"]
+        assert_equal 1 [llength [dict get $user selectors]]
+        assert_equal "" [dict get [lindex [dict get $user selectors] 0] keys]
+        assert_equal "&*" [dict get [lindex [dict get $user selectors] 0] channels]
+        assert_equal "-@all" [dict get [lindex [dict get $user selectors] 0] commands]
+    }
+
+    test {Test deleting selectors} {
+        r ACL SETUSER selector-del on "(~added-selector)"
+        set user [r ACL GETUSER "selector-del"]
+        assert_equal "~added-selector" [dict get [lindex [dict get $user selectors] 0] keys]
+        assert_equal [llength [dict get $user selectors]] 1
+
+        r ACL SETUSER selector-del clearselectors
+        set user [r ACL GETUSER "selector-del"]
+        assert_equal [llength [dict get $user selectors]] 0
+    }
+
+    test {Test selector syntax error reports the error in the selector context} {
+        catch {r ACL SETUSER selector-syntax on (this-is-invalid)} e
+        assert_match "*ERR Error in ACL SETUSER modifier '(*)*Syntax*" $e
+
+        catch {r ACL SETUSER selector-syntax on (&fail)} e
+        assert_match "*ERR Error in ACL SETUSER modifier '(*)*Adding a pattern after the*" $e
+
+        assert_equal "" [r ACL GETUSER selector-syntax]
+    }
+
+    test {Test flexible selector definition} {
+        # Test valid selectors
+        r ACL SETUSER selector-2 "(~key1 +get )" "( ~key2 +get )" "( ~key3 +get)" "(~key4 +get)"
+        r ACL SETUSER selector-2 (~key5 +get ) ( ~key6 +get ) ( ~key7 +get) (~key8 +get)
+        set user [r ACL GETUSER "selector-2"]
+        assert_equal "~key1" [dict get [lindex [dict get $user selectors] 0] keys]
+        assert_equal "~key2" [dict get [lindex [dict get $user selectors] 1] keys]
+        assert_equal "~key3" [dict get [lindex [dict get $user selectors] 2] keys]
+        assert_equal "~key4" [dict get [lindex [dict get $user selectors] 3] keys]
+        assert_equal "~key5" [dict get [lindex [dict get $user selectors] 4] keys]
+        assert_equal "~key6" [dict get [lindex [dict get $user selectors] 5] keys]
+        assert_equal "~key7" [dict get [lindex [dict get $user selectors] 6] keys]
+        assert_equal "~key8" [dict get [lindex [dict get $user selectors] 7] keys]
+
+        # Test invalid selector syntax
+        catch {r ACL SETUSER invalid-selector " () "} err
+        assert_match "*ERR*Syntax error*" $err
+        catch {r ACL SETUSER invalid-selector (} err
+        assert_match "*Unmatched parenthesis*" $err
+        catch {r ACL SETUSER invalid-selector )} err
+        assert_match "*ERR*Syntax error" $err
+    }
+
+    test {Test separate read permission} {
+        r ACL SETUSER key-permission-R on nopass %R~read* +@all
+        $r2 auth key-permission-R password
+        assert_equal PONG [$r2 PING]
+        r set readstr bar
+        assert_equal bar [$r2 get readstr]
+        catch {$r2 set readstr bar} err
+        assert_match "*NOPERM*keys*" $err
+        catch {$r2 get notread} err
+        assert_match "*NOPERM*keys*" $err
+    }
+
+    test {Test separate write permission} {
+        r ACL SETUSER key-permission-W on nopass %W~write* +@all
+        $r2 auth key-permission-W password
+        assert_equal PONG [$r2 PING]
+        # Note, SET is a RW command, so it's not used for testing
+        $r2 LPUSH writelist 10
+        catch {$r2 GET writestr} err
+        assert_match "*NOPERM*keys*" $err
+        catch {$r2 LPUSH notwrite 10} err
+        assert_match "*NOPERM*keys*" $err
+    }
+
+    test {Test separate read and write permissions} {
+        r ACL SETUSER key-permission-RW on nopass %R~read* %W~write* +@all
+        $r2 auth key-permission-RW password
+        assert_equal PONG [$r2 PING]
+        r set read bar
+        $r2 copy read write
+        catch {$r2 copy write read} err
+        assert_match "*NOPERM*keys*" $err        
+    }
+
+    test {Test separate read and write permissions on different selectors are not additive} {
+        r ACL SETUSER key-permission-RW-selector on nopass "(%R~read* +@all)" "(%W~write* +@all)"
+        $r2 auth key-permission-RW-selector password
+        assert_equal PONG [$r2 PING]
+
+        # Verify write selector
+        $r2 LPUSH writelist 10
+        catch {$r2 GET writestr} err
+        assert_match "*NOPERM*keys*" $err
+        catch {$r2 LPUSH notwrite 10} err
+        assert_match "*NOPERM*keys*" $err
+
+        # Verify read selector
+        r set readstr bar
+        assert_equal bar [$r2 get readstr]
+        catch {$r2 set readstr bar} err
+        assert_match "*NOPERM*keys*" $err
+        catch {$r2 get notread} err
+        assert_match "*NOPERM*keys*" $err
+
+        # Verify they don't combine
+        catch {$r2 copy read write} err
+        assert_match "*NOPERM*keys*" $err
+        catch {$r2 copy write read} err
+        assert_match "*NOPERM*keys*" $err   
+    }
+
+    test {Test ACL log correctly identifies the relevant item when selectors are used} {
+        r ACL SETUSER acl-log-test-selector on nopass 
+        r ACL SETUSER acl-log-test-selector +mget ~key (+mget ~key ~otherkey)
+        $r2 auth acl-log-test-selector password
+
+        # Test that command is shown only if none of the selectors match
+        r ACL LOG RESET
+        catch {$r2 GET key} err
+        assert_match "*NOPERM*command*" $err
+        set entry [lindex [r ACL LOG] 0]
+        assert_equal [dict get $entry username] "acl-log-test-selector"
+        assert_equal [dict get $entry context] "toplevel"
+        assert_equal [dict get $entry reason] "command"
+        assert_equal [dict get $entry object] "get"
+
+        # Test two cases where the first selector matches less than the
+        # second selector. We should still show the logically first unmatched key.
+        r ACL LOG RESET
+        catch {$r2 MGET otherkey someotherkey} err
+        assert_match "*NOPERM*keys*" $err
+        set entry [lindex [r ACL LOG] 0]
+        assert_equal [dict get $entry username] "acl-log-test-selector"
+        assert_equal [dict get $entry context] "toplevel"
+        assert_equal [dict get $entry reason] "key"
+        assert_equal [dict get $entry object] "someotherkey"
+
+        r ACL LOG RESET
+        catch {$r2 MGET key otherkey someotherkey} err
+        assert_match "*NOPERM*keys*" $err
+        set entry [lindex [r ACL LOG] 0]
+        assert_equal [dict get $entry username] "acl-log-test-selector"
+        assert_equal [dict get $entry context] "toplevel"
+        assert_equal [dict get $entry reason] "key"
+        assert_equal [dict get $entry object] "someotherkey"
+    }
+
+    test {Test ACL GETUSER response information} {
+        r ACL setuser selector-info -@all +get resetchannels &channel1 %R~foo1 %W~bar1 ~baz1 
+        r ACL setuser selector-info (-@all +set resetchannels &channel2 %R~foo2 %W~bar2 ~baz2)
+        set user [r ACL GETUSER "selector-info"]
+    
+        # Root selector
+        assert_equal "%R~foo1 %W~bar1 ~baz1" [dict get $user keys]
+        assert_equal "&channel1" [dict get $user channels]
+        assert_equal "-@all +get" [dict get $user commands]
+
+        # Added selector
+        set secondary_selector [lindex [dict get $user selectors] 0]
+        assert_equal "%R~foo2 %W~bar2 ~baz2" [dict get $secondary_selector keys]
+        assert_equal "&channel2" [dict get $secondary_selector channels]
+        assert_equal "-@all +set" [dict get $secondary_selector commands] 
+    }
+
+    test {Test ACL list idempotency} {
+        r ACL SETUSER user-idempotency off -@all +get resetchannels &channel1 %R~foo1 %W~bar1 ~baz1 (-@all +set resetchannels &channel2 %R~foo2 %W~bar2 ~baz2)
+        set response [lindex [r ACL LIST] [lsearch [r ACL LIST] "user user-idempotency*"]]
+
+        assert_match "*-@all*+get*(*)*" $response
+        assert_match "*resetchannels*&channel1*(*)*" $response
+        assert_match "*%R~foo1*%W~bar1*~baz1*(*)*" $response
+
+        assert_match "*(*-@all*+set*)*" $response
+        assert_match "*(*resetchannels*&channel2*)*" $response
+        assert_match "*(*%R~foo2*%W~bar2*~baz2*)*" $response
+    }
+
+    test {Test R+W is the same as all permissions} {
+        r ACL setuser selector-rw-info %R~foo %W~foo %RW~bar
+        set user [r ACL GETUSER selector-rw-info]
+        assert_equal "~foo ~bar" [dict get $user keys]
+    }
+
+    test {Test basic dry run functionality} {
+        r ACL setuser command-test +@all %R~read* %W~write* %RW~rw*
+        assert_equal "OK" [r ACL DRYRUN command-test GET read]
+
+        catch {r ACL DRYRUN not-a-user GET read} e
+        assert_equal "ERR User 'not-a-user' not found" $e
+
+        catch {r ACL DRYRUN command-test not-a-command read} e
+        assert_equal "ERR Command 'not-a-command' not found" $e
+    }
+
+    test {Test various odd commands for key permissions} {
+        r ACL setuser command-test +@all %R~read* %W~write* %RW~rw*
+
+        # Test migrate, which is marked with incomplete keys
+        assert_equal "OK" [r ACL DRYRUN command-test MIGRATE whatever whatever rw]
+        assert_equal "This user has no permissions to access the 'read' key" [r ACL DRYRUN command-test MIGRATE whatever whatever read]
+        assert_equal "This user has no permissions to access the 'write' key" [r ACL DRYRUN command-test MIGRATE whatever whatever write]
+        assert_equal "OK" [r ACL DRYRUN command-test MIGRATE whatever whatever "" 0 5000 KEYS rw]
+        assert_equal "This user has no permissions to access the 'read' key" [r ACL DRYRUN command-test MIGRATE whatever whatever "" 0 5000 KEYS read]
+        assert_equal "This user has no permissions to access the 'write' key" [r ACL DRYRUN command-test MIGRATE whatever whatever "" 0 5000 KEYS write]
+
+        # Test SORT, which is marked with incomplete keys
+        assert_equal "OK" [r ACL DRYRUN command-test SORT read STORE write]
+        assert_equal "This user has no permissions to access the 'read' key"  [r ACL DRYRUN command-test SORT read STORE read]
+        assert_equal "This user has no permissions to access the 'write' key"  [r ACL DRYRUN command-test SORT write STORE write]
+
+        # Test EVAL, which uses the numkey keyspec (Also test EVAL_RO)
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL "" 1 rw1]
+        assert_equal "This user has no permissions to access the 'read' key" [r ACL DRYRUN command-test EVAL "" 1 read]
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL_RO "" 1 rw1]
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL_RO "" 1 read]
+
+        # Read is an optional argument and not a key here, make sure we don't treat it as a key
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL "" 0 read]
+
+        # These are syntax errors, but it's 'OK' from an ACL perspective
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL "" -1 read]
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL "" 3 rw rw]
+        assert_equal "OK" [r ACL DRYRUN command-test EVAL "" 3 rw read]
+
+        # Test GEORADIUS which uses the last type of keyspec, keyword
+        assert_equal "OK" [r ACL DRYRUN command-test GEORADIUS read longitude latitude radius M STOREDIST write]
+        assert_equal "OK" [r ACL DRYRUN command-test GEORADIUS read longitude latitude radius M]
+        assert_equal "This user has no permissions to access the 'read2' key" [r ACL DRYRUN command-test GEORADIUS read1 longitude latitude radius M STOREDIST read2]
+        assert_equal "This user has no permissions to access the 'write1' key" [r ACL DRYRUN command-test GEORADIUS write1 longitude latitude radius M STOREDIST write2]
+        assert_equal "OK" [r ACL DRYRUN command-test GEORADIUS read longitude latitude radius M STORE write]
+        assert_equal "OK" [r ACL DRYRUN command-test GEORADIUS read longitude latitude radius M]
+        assert_equal "This user has no permissions to access the 'read2' key" [r ACL DRYRUN command-test GEORADIUS read1 longitude latitude radius M STORE read2]
+        assert_equal "This user has no permissions to access the 'write1' key" [r ACL DRYRUN command-test GEORADIUS write1 longitude latitude radius M STORE write2]
+    }
+
+    test {Test sharded channel permissions} {
+        r ACL setuser test-channels +@all resetchannels &channel
+        assert_equal "OK" [r ACL DRYRUN test-channels spublish channel foo]
+        assert_equal "OK" [r ACL DRYRUN test-channels ssubscribe channel]
+        assert_equal "OK" [r ACL DRYRUN test-channels sunsubscribe]
+        assert_equal "OK" [r ACL DRYRUN test-channels sunsubscribe channel]
+        assert_equal "OK" [r ACL DRYRUN test-channels sunsubscribe otherchannel]
+
+        assert_equal "This user has no permissions to access the 'otherchannel' channel" [r ACL DRYRUN test-channels spublish otherchannel foo]
+        assert_equal "This user has no permissions to access the 'otherchannel' channel" [r ACL DRYRUN test-channels ssubscribe otherchannel foo]
+    }
+
+    $r2 close
+}
+
+set server_path [tmpdir "selectors.acl"]
+exec cp -f tests/assets/userwithselectors.acl $server_path
+exec cp -f tests/assets/default.conf $server_path
+start_server [list overrides [list "dir" $server_path "aclfile" "userwithselectors.acl"] tags [list "external:skip"]] {
+
+    test {Test behavior of loading ACLs} {
+        set selectors [dict get [r ACL getuser alice] selectors]
+        assert_equal [llength $selectors] 1
+        set test_selector [lindex $selectors 0]
+        assert_equal "-@all +get" [dict get $test_selector "commands"]
+        assert_equal "~rw*" [dict get $test_selector "keys"]
+
+        set selectors [dict get [r ACL getuser bob] selectors]
+        assert_equal [llength $selectors] 2
+        set test_selector [lindex $selectors 0]
+        assert_equal "-@all +set" [dict get $test_selector "commands"]
+        assert_equal "%W~w*" [dict get $test_selector "keys"]
+
+        set test_selector [lindex $selectors 1]
+        assert_equal "-@all +get" [dict get $test_selector "commands"]
+        assert_equal "%R~r*" [dict get $test_selector "keys"]
+    }
+}
diff --git a/tests/unit/acl.tcl b/tests/unit/acl.tcl
index 87f6fb46a..872bade90 100644
--- a/tests/unit/acl.tcl
+++ b/tests/unit/acl.tcl
@@ -733,27 +733,27 @@ start_server [list overrides [list "dir" $server_path "acl-pubsub-default" "rese
 
     test {Default user has access to all channels irrespective of flag} {
         set channelinfo [dict get [r ACL getuser default] channels]
-        assert_equal "*" $channelinfo
+        assert_equal "&*" $channelinfo
         set channelinfo [dict get [r ACL getuser alice] channels]
         assert_equal "" $channelinfo
     }
 
     test {Update acl-pubsub-default, existing users shouldn't get affected} {
         set channelinfo [dict get [r ACL getuser default] channels]
-        assert_equal "*" $channelinfo
+        assert_equal "&*" $channelinfo
         r CONFIG set acl-pubsub-default allchannels
         r ACL setuser mydefault
         set channelinfo [dict get [r ACL getuser mydefault] channels]
-        assert_equal "*" $channelinfo
+        assert_equal "&*" $channelinfo
         r CONFIG set acl-pubsub-default resetchannels
         set channelinfo [dict get [r ACL getuser mydefault] channels]
-        assert_equal "*" $channelinfo
+        assert_equal "&*" $channelinfo
     }
 
     test {Single channel is valid} {
         r ACL setuser onechannel &test
         set channelinfo [dict get [r ACL getuser onechannel] channels]
-        assert_equal test $channelinfo
+        assert_equal "&test" $channelinfo
         r ACL deluser onechannel
     }
 
@@ -772,7 +772,7 @@ start_server [list overrides [list "dir" $server_path "acl-pubsub-default" "rese
 
     test {Only default user has access to all channels irrespective of flag} {
         set channelinfo [dict get [r ACL getuser default] channels]
-        assert_equal "*" $channelinfo
+        assert_equal "&*" $channelinfo
         set channelinfo [dict get [r ACL getuser alice] channels]
         assert_equal "" $channelinfo
     }
diff --git a/tests/unit/moduleapi/aclcheck.tcl b/tests/unit/moduleapi/aclcheck.tcl
index 94af2cc00..3fa3ed43f 100644
--- a/tests/unit/moduleapi/aclcheck.tcl
+++ b/tests/unit/moduleapi/aclcheck.tcl
@@ -20,11 +20,20 @@ start_server {tags {"modules acl"}} {
 
     test {test module check acl for key perm} {
         # give permission for SET and block all keys but x
-        r acl setuser default +set resetkeys ~x
-        assert_equal [r aclcheck.set.check.key x 5] OK
-        catch {r aclcheck.set.check.key y 5} e
-        set e
-    } {*DENIED KEY*}
+        r acl setuser default +set resetkeys ~x %W~y %R~z
+
+        assert_equal [r aclcheck.set.check.key "*" x 5] OK
+        catch {r aclcheck.set.check.key "*" v 5} e
+        assert_match "*DENIED KEY*" $e
+
+        assert_equal [r aclcheck.set.check.key "W" y 5] OK
+        catch {r aclcheck.set.check.key "W" v 5} e
+        assert_match "*DENIED KEY*" $e
+
+        assert_equal [r aclcheck.set.check.key "R" z 5] OK
+        catch {r aclcheck.set.check.key "R" v 5} e
+        assert_match "*DENIED KEY*" $e
+    }
 
     test {test module check acl for module user} {
         # the module user has access to all keys