Adds pub/sub channel patterns to ACL (#7993)

Fixes #7923.

This PR appropriates the special `&` symbol (because `@` and `*` are taken),
followed by a literal value or pattern for describing the Pub/Sub patterns that
an ACL user can interact with. It is similar to the existing key patterns
mechanism in function (additive) and implementation (copy-pasta). It also adds
the allchannels and resetchannels ACL keywords, naturally.

The default user is given allchannels permissions, whereas new users get
whatever is defined by the acl-pubsub-default configuration directive. For
backward compatibility in 6.2, the default of this directive is allchannels but
this is likely to be changed to resetchannels in the next major version for
stronger default security settings.

Unless allchannels is set for the user, channel access permissions are checked
as follows :
* Calls to both PUBLISH and SUBSCRIBE will fail unless a pattern matching the
  argumentative channel name(s) exists for the user.
* Calls to PSUBSCRIBE will fail unless the pattern(s) provided as an argument
  literally exist(s) in the user's list.

Such failures are logged to the ACL log.

Runtime changes to channel permissions for a user with existing subscribing
clients cause said clients to disconnect unless the new permissions permit the
connections to continue. Note, however, that PSUBSCRIBErs' patterns are matched
literally, so given the change bar:* -> b*, pattern subscribers to bar:* will be
disconnected.

Notes/questions:
* UNSUBSCRIBE, PUNSUBSCRIBE and PUBSUB remain unprotected due to lack of reasons
  for touching them.

redis.conf

@@ -747,6 +747,11 @@ replica-priority 100
 #               It is possible to specify multiple patterns.
 #  allkeys      Alias for ~*
 #  resetkeys    Flush the list of allowed keys patterns.
+#  &<pattern>   Add a glob-style pattern of Pub/Sub channels that can be
+#               accessed by the user. It is possible to specify multiple channel
+#               patterns.
+#  allchannels  Alias for &*
+#  resetchannels            Flush the list of allowed channel patterns.
 #  ><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).
@@ -820,6 +825,27 @@ acllog-max-len 128
 #
 # requirepass foobared
 
+# New users are initialized with restrictive permissions by default, via the
+# equivalent of this ACL rule 'off resetkeys -@all'. Starting with Redis 6.2, it
+# is possible to manage access to Pub/Sub channels with ACL rules as well. The
+# default Pub/Sub channels permission if new users is controlled by the 
+# acl-pubsub-default configuration directive, which accepts one of these values:
+#
+# allchannels: grants access to all Pub/Sub channels
+# resetchannels: revokes access to all Pub/Sub channels
+#
+# To ensure backward compatibility while upgrading Redis 6.0, acl-pubsub-default
+# defaults to the 'allchannels' permission.
+#
+# Future compatibility note: it is very likely that in a future version of Redis
+# the directive's default of 'allchannels' will be changed to 'resetchannels' in
+# order to provide better out-of-the-box Pub/Sub security. Therefore, it is
+# recommended that you explicitly define Pub/Sub permissions for all users
+# rather then rely on implicit default values. Once you've set explicit
+# Pub/Sub for all exisitn users, you should uncomment the following line.
+#
+# acl-pubsub-default resetchannels
+
 # Command renaming (DEPRECATED).
 #
 # ------------------------------------------------------------------------

src/acl.c

@@ -92,6 +92,7 @@ struct ACLUserFlag {
     {"on", USER_FLAG_ENABLED},
     {"off", USER_FLAG_DISABLED},
     {"allkeys", USER_FLAG_ALLKEYS},
+    {"allchannels", USER_FLAG_ALLCHANNELS},
     {"allcommands", USER_FLAG_ALLCOMMANDS},
     {"nopass", USER_FLAG_NOPASS},
     {NULL,0} /* Terminator. */
@@ -241,16 +242,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;
+    u->flags = USER_FLAG_DISABLED | server.acl_pubusub_default;
     u->allowed_subcommands = NULL;
     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));
     raxInsert(Users,(unsigned char*)name,namelen,u,NULL);
     return u;
@@ -279,6 +284,7 @@ void ACLFreeUser(user *u) {
     sdsfree(u->name);
     listRelease(u->passwords);
     listRelease(u->patterns);
+    listRelease(u->channels);
     ACLResetSubcommands(u);
     zfree(u);
 }
@@ -319,8 +325,10 @@ void ACLFreeUserAndKillClients(user *u) {
 void ACLCopyUser(user *dst, user *src) {
     listRelease(dst->passwords);
     listRelease(dst->patterns);
+    listRelease(dst->channels);
     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->flags = src->flags;
@@ -602,9 +610,10 @@ sds ACLDescribeUser(user *u) {
 
     /* Flags. */
     for (int j = 0; ACLUserFlags[j].flag; j++) {
-        /* Skip the allcommands and allkeys flags because they'll be emitted
+        /* Skip the allcommands, allkeys and allchannels flags because they'll
-         * later as ~* and +@all. */
+         * 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);
@@ -636,6 +645,19 @@ sds ACLDescribeUser(user *u) {
         }
     }
 
+    /* Pub/sub channel patterns. */
+    if (u->flags & USER_FLAG_ALLCHANNELS) {
+        res = sdscatlen(res,"&* ",3);
+    } else {
+        listRewind(u->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 = ACLDescribeUserCommandRules(u);
     res = sdscatsds(res,rules);
@@ -681,7 +703,6 @@ void ACLResetSubcommands(user *u) {
     u->allowed_subcommands = NULL;
 }
 
-
 /* Add a subcommand to the list of subcommands for the user 'u' and
  * the command id specified. */
 void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
@@ -742,6 +763,12 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  *              It is possible to specify multiple patterns.
  * allkeys      Alias for ~*
  * resetkeys    Flush the list of allowed keys patterns.
+ * &<pattern>   Add a pattern of channels that can be mentioned as part of
+ *              Pub/Sub commands. For instance &* allows all the channels. The
+ *              pattern is a glob-style pattern like the one of PSUBSCRIBE.
+ *              It is possible to specify multiple patterns.
+ * allchannels              Alias for &*
+ * resetchannels            Flush the list of allowed keys patterns.
  * ><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).
@@ -780,7 +807,7 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  *
  * When an error is returned, errno is set to the following values:
  *
- * EINVAL: The specified opcode is not understood or the key pattern is
+ * EINVAL: The specified opcode is not understood or the key/channel pattern is
  *         invalid (contains non allowed characters).
  * ENOENT: The command name or command category provided with + or - is not
  *         known.
@@ -788,6 +815,8 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  *         fully added.
  * EEXIST: You are adding a key pattern after "*" was already added. This is
  *         almost surely an error on the user side.
+ * EISDIR: You are adding a channel pattern after "*" was already added. This is
+ *         almost surely an error on the user side.
  * ENODEV: The password you are trying to remove from the user does not exist.
  * EBADMSG: The hash you are trying to add is not a valid hash.
  */
@@ -808,6 +837,14 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     } 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"))
     {
@@ -875,12 +912,29 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
         }
         sds newpat = sdsnewlen(op+1,oplen-1);
         listNode *ln = listSearchKey(u->patterns,newpat);
-        /* Avoid re-adding the same pattern multiple times. */
+        /* 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 (strchr(op,'|') == NULL) {
             if (ACLLookupCommand(op+1) == NULL) {
@@ -948,6 +1002,7 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     } 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);
         serverAssert(ACLSetUser(u,"off",-1) == C_OK);
         serverAssert(ACLSetUser(u,"-@all",-1) == C_OK);
     } else {
@@ -974,6 +1029,11 @@ char *ACLSetUserStringError(void) {
                  "'allkeys' flag) is not valid and does not have any "
                  "effect. Try 'resetkeys' to start with an empty "
                  "list of patterns";
+    else if (errno == EISDIR)
+        errmsg = "Adding a pattern after the * pattern (or the "
+                 "'allchannels' flag) is not valid and does not have any "
+                 "effect. Try 'resetchannels' to start with an empty "
+                 "list of channels";
     else if (errno == ENODEV)
         errmsg = "The password you are trying to remove from the user does "
                  "not exist";
@@ -989,6 +1049,7 @@ void ACLInitDefaultUser(void) {
     DefaultUser = ACLCreateUser("default",7);
     ACLSetUser(DefaultUser,"+@all",-1);
     ACLSetUser(DefaultUser,"~*",-1);
+    ACLSetUser(DefaultUser,"&*",-1);
     ACLSetUser(DefaultUser,"on",-1);
     ACLSetUser(DefaultUser,"nopass",-1);
 }
@@ -1193,6 +1254,119 @@ int ACLCheckCommandPerm(client *c, int *keyidxptr) {
     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) {
+    listIter li;
+    listNode *ln;
+    size_t clen = sdslen(channel);
+    int match = 0;
+
+    listRewind(allowed,&li);
+    while((ln = listNext(&li))) {
+        sds pattern = listNodeValue(ln);
+        size_t plen = sdslen(pattern);
+
+        if ((literal && !sdscmp(pattern,channel)) || 
+            (!literal && stringmatchlen(pattern,plen,channel,clen,0)))
+        {
+            match = 1;
+            break;
+        }
+    }
+    if (!match) {
+        return ACL_DENIED_CHANNEL;
+    }
+    return ACL_OK;
+}
+
+/* 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) {
+    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
+     * permissions. */
+    listRewind(u->channels,&li);
+    while (!kill && ((ln = listNext(&li)) != NULL)) {
+        sds pattern = listNodeValue(ln);
+        kill = (ACLCheckPubsubChannelPerm(pattern,upcoming,1) ==
+                ACL_DENIED_CHANNEL);
+    }
+    if (!kill) return;
+
+    /* 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) {
+            /* 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);
+            }
+            /* Check for channel violations. */
+            if (!kill) {
+                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);
+                }
+                dictReleaseIterator(di);
+            }
+
+            /* Kill it. */
+            if (kill) {
+                freeClient(c);
+            }
+        }
+    }
+}
+
+/* Check if the pub/sub channels of the command, that's ready to be executed in
+ * the client 'c', can be executed by this client according to the ACLs channels
+ * associated to the client user c->user.
+ * 
+ * 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(client *c, int idx, int count, int literal, int *idxptr) {
+    user *u = c->user;
+
+    /* If there is no associated user, the connection can run anything. */
+    if (u == NULL) return ACL_OK;
+
+    /* Check if the user can access the channels mentioned in the command's
+     * arguments. */
+    if (!(c->user->flags & USER_FLAG_ALLCHANNELS)) {
+        for (int j = idx; j < idx+count; j++) {
+            if (ACLCheckPubsubChannelPerm(c->argv[j]->ptr,u->channels,literal)
+                != ACL_OK) {
+                if (idxptr) *idxptr = j;
+                return ACL_DENIED_CHANNEL;
+            }
+        }
+    }
+
+    /* If we survived all the above checks, the user can execute the
+     * command. */
+    return ACL_OK;
+
+}
+
 /* =============================================================================
  * ACL loading / saving functions
  * ==========================================================================*/
@@ -1608,13 +1782,13 @@ void ACLFreeLogEntry(void *leptr) {
  * the log entry instead of creating many entries for very similar ACL
  * rules issues.
  *
- * The keypos argument is only used when the reason is ACL_DENIED_KEY, since
+ * The argpos argument is used when the reason is ACL_DENIED_KEY or 
- * it allows the function to log the key name that caused the problem.
+ * ACL_DENIED_CHANNEL, since it allows the function to log the key or channel
- * Similarly the username is only passed when we failed to authenticate the
+ * name that caused the problem. Similarly the username is only passed when we
- * user with AUTH or HELLO, for the ACL_DENIED_AUTH reason. Otherwise
+ * failed to authenticate the user with AUTH or HELLO, for the ACL_DENIED_AUTH
- * it will just be NULL.
+ * reason. Otherwise it will just be NULL.
  */
-void addACLLogEntry(client *c, int reason, int keypos, sds username) {
+void addACLLogEntry(client *c, int reason, int argpos, sds username) {
     /* Create a new entry. */
     struct ACLLogEntry *le = zmalloc(sizeof(*le));
     le->count = 1;
@@ -1624,8 +1798,9 @@ void addACLLogEntry(client *c, int reason, int keypos, sds username) {
 
     switch(reason) {
     case ACL_DENIED_CMD: le->object = sdsnew(c->cmd->name); break;
-    case ACL_DENIED_KEY: le->object = sdsnew(c->argv[keypos]->ptr); break;
+    case ACL_DENIED_KEY: le->object = sdsdup(c->argv[argpos]->ptr); break;
-    case ACL_DENIED_AUTH: le->object = sdsnew(c->argv[0]->ptr); break;
+    case ACL_DENIED_CHANNEL: le->object = sdsdup(c->argv[argpos]->ptr); break;
+    case ACL_DENIED_AUTH: le->object = sdsdup(c->argv[0]->ptr); break;
     default: le->object = sdsempty();
     }
 
@@ -1733,6 +1908,11 @@ void aclCommand(client *c) {
             }
         }
 
+        /* 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);
+
         /* Overwrite the user with the temporary user we modified above. */
         if (!u) u = ACLCreateUser(username,sdslen(username));
         serverAssert(u != NULL);
@@ -1768,7 +1948,7 @@ void aclCommand(client *c) {
             return;
         }
 
-        addReplyMapLen(c,4);
+        addReplyMapLen(c,5);
 
         /* Flags */
         addReplyBulkCString(c,"flags");
@@ -1813,6 +1993,22 @@ void aclCommand(client *c) {
                 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));
+            }
+        }
     } else if ((!strcasecmp(sub,"list") || !strcasecmp(sub,"users")) &&
                c->argc == 2)
     {
@@ -1950,6 +2146,7 @@ void aclCommand(client *c) {
             switch(le->reason) {
             case ACL_DENIED_CMD: reasonstr="command"; break;
             case ACL_DENIED_KEY: reasonstr="key"; break;
+            case ACL_DENIED_CHANNEL: reasonstr="channel"; break;
             case ACL_DENIED_AUTH: reasonstr="auth"; break;
             default: reasonstr="unknown";
             }

src/config.c

@@ -113,6 +113,12 @@ configEnum oom_score_adj_enum[] = {
     {NULL, 0}
 };
 
+configEnum acl_pubsub_default_enum[] = {
+    {"allchannels", USER_FLAG_ALLCHANNELS},
+    {"resetchannels", 0},
+    {NULL, 0}
+};
+
 /* Output buffer limits presets. */
 clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUNT] = {
     {0, 0, 0}, /* normal */
@@ -2352,6 +2358,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_pubusub_default, USER_FLAG_ALLCHANNELS, NULL, NULL),
 
     /* Integer configs */
     createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.dbnum, 16, INTEGER_CONFIG, NULL, NULL),

src/multi.c

@@ -198,18 +198,34 @@ void execCommand(client *c) {
             must_propagate = 1;
         }
 
-        int acl_keypos;
+        /* ACL permissions are also checked at the time of execution in case
-        int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
+         * they were changed after the commands were ququed. */
+        int acl_errpos;
+        int acl_retval = ACLCheckCommandPerm(c,&acl_errpos);
+        if (acl_retval == ACL_OK && c->cmd->proc == publishCommand)
+            acl_retval = ACLCheckPubsubPerm(c,1,1,0,&acl_errpos);
         if (acl_retval != ACL_OK) {
-            addACLLogEntry(c,acl_retval,acl_keypos,NULL);
+            char *reason;
+            switch (acl_retval) {
+            case ACL_DENIED_CMD:
+                reason = "no permission to execute the command or subcommand";
+                break;
+            case ACL_DENIED_KEY:
+                reason = "no permission to touch the specified keys";
+                break;
+            case ACL_DENIED_CHANNEL:
+                reason = "no permission to publish to the specified channel";
+                break;
+            default:
+                reason = "no permission";
+                break;
+            }
+            addACLLogEntry(c,acl_retval,acl_errpos,NULL);
             addReplyErrorFormat(c,
                 "-NOPERM ACLs rules changed between the moment the "
                 "transaction was accumulated and the EXEC call. "
                 "This command is no longer allowed for the "
-                "following reason: %s",
+                "following reason: %s", reason);
-                (acl_retval == ACL_DENIED_CMD) ?
-                "no permission to execute the command or subcommand" :
-                "no permission to touch the specified keys");
         } else {
             call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL);
             serverAssert((c->flags & CLIENT_BLOCKED) == 0);

src/pubsub.c

@@ -353,13 +353,29 @@ int pubsubPublishMessage(robj *channel, robj *message) {
     return receivers;
 }
 
+/* This wraps handling ACL channel permissions for the given client. */
+int pubsubCheckACLPermissionsOrReply(client *c, int idx, int count, int literal) {
+    /* Check if the user can run the command according to the current
+     * ACLs. */
+    int acl_chanpos;
+    int acl_retval = ACLCheckPubsubPerm(c,idx,count,literal,&acl_chanpos);
+    if (acl_retval == ACL_DENIED_CHANNEL) {
+        addACLLogEntry(c,acl_retval,acl_chanpos,NULL);
+        addReplyError(c,
+            "-NOPERM this user has no permissions to access "
+            "one of the channels used as arguments");
+    }
+    return acl_retval;
+}
+
 /*-----------------------------------------------------------------------------
  * Pubsub commands implementation
  *----------------------------------------------------------------------------*/
 
+/* SUBSCRIBE channel [channel ...] */
 void subscribeCommand(client *c) {
     int j;
-
+    if (pubsubCheckACLPermissionsOrReply(c,1,c->argc-1,0) != ACL_OK) return;
     if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
         /**
          * A client that has CLIENT_DENY_BLOCKING flag on
@@ -368,7 +384,7 @@ void subscribeCommand(client *c) {
          * Notice that we have a special treatment for multi because of
          * backword compatibility
          */
-        addReplyError(c, "subscribe is not allow on DENY BLOCKING client");
+        addReplyError(c, "SUBSCRIBE isn't allowed for a DENY BLOCKING client");
         return;
     }
 
@@ -377,6 +393,7 @@ void subscribeCommand(client *c) {
     c->flags |= CLIENT_PUBSUB;
 }
 
+/* UNSUBSCRIBE [channel [channel ...]] */
 void unsubscribeCommand(client *c) {
     if (c->argc == 1) {
         pubsubUnsubscribeAllChannels(c,1);
@@ -389,9 +406,10 @@ void unsubscribeCommand(client *c) {
     if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
 }
 
+/* PSUBSCRIBE pattern [pattern ...] */
 void psubscribeCommand(client *c) {
     int j;
-
+    if (pubsubCheckACLPermissionsOrReply(c,1,c->argc-1,1) != ACL_OK) return;
     if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
         /**
          * A client that has CLIENT_DENY_BLOCKING flag on
@@ -400,7 +418,7 @@ void psubscribeCommand(client *c) {
          * Notice that we have a special treatment for multi because of
          * backword compatibility
          */
-        addReplyError(c, "PSUBSCRIBE is not allowed for DENY BLOCKING client");
+        addReplyError(c, "PSUBSCRIBE isn't allowed for a DENY BLOCKING client");
         return;
     }
 
@@ -409,6 +427,7 @@ void psubscribeCommand(client *c) {
     c->flags |= CLIENT_PUBSUB;
 }
 
+/* PUNSUBSCRIBE [pattern [pattern ...]] */
 void punsubscribeCommand(client *c) {
     if (c->argc == 1) {
         pubsubUnsubscribeAllPatterns(c,1);
@@ -421,7 +440,9 @@ void punsubscribeCommand(client *c) {
     if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
 }
 
+/* PUBLISH <channel> <message> */
 void publishCommand(client *c) {
+    if (pubsubCheckACLPermissionsOrReply(c,1,1,0) != ACL_OK) return;
     int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
     if (server.cluster_enabled)
         clusterPropagatePublish(c->argv[1],c->argv[2]);

src/redis-cli.c

@@ -1363,9 +1363,16 @@ static int cliSendCommand(int argc, char **argv, long repeat) {
             /* Unset our default PUSH handler so this works in RESP2/RESP3 */
             redisSetPushCallback(context, NULL);
 
-            while (1) {
+            while (config.pubsub_mode) {
                 if (cliReadReply(output_raw) != REDIS_OK) exit(1);
+                if (config.last_cmd_type == REDIS_REPLY_ERROR) {
+                    if (config.push_output) {
+                        redisSetPushCallback(context, cliPushHandler);
+                    }
+                    config.pubsub_mode = 0;
+                }
             }
+            continue;
         }
 
         if (config.slave_mode) {

src/scripting.c

@@ -606,17 +606,31 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     }
 
     /* Check the ACLs. */
-    int acl_keypos;
+    int acl_errpos;
-    int acl_retval = ACLCheckCommandPerm(c,&acl_keypos);
+    int acl_retval = ACLCheckCommandPerm(c,&acl_errpos);
+    if (acl_retval == ACL_OK && c->cmd->proc == publishCommand)
+        acl_retval = ACLCheckPubsubPerm(c,1,1,0,&acl_errpos);
     if (acl_retval != ACL_OK) {
-        addACLLogEntry(c,acl_retval,acl_keypos,NULL);
+        addACLLogEntry(c,acl_retval,acl_errpos,NULL);
-        if (acl_retval == ACL_DENIED_CMD)
+        switch (acl_retval) {
+        case ACL_DENIED_CMD:
             luaPushError(lua, "The user executing the script can't run this "
                               "command or subcommand");
-        else
+            break;
+        case ACL_DENIED_KEY:
             luaPushError(lua, "The user executing the script can't access "
                               "at least one of the keys mentioned in the "
                               "command arguments");
+            break;
+        case ACL_DENIED_AUTH:
+            luaPushError(lua, "The user executing the script can't publish "
+                              "to the channel mentioned in the command");
+            break;
+        default:
+            luaPushError(lua, "The user executing the script is lacking the "
+                              "permissions for the command");
+            break;
+        }
         goto cleanup;
     }
 

src/server.c

@@ -4899,7 +4899,7 @@ void monitorCommand(client *c) {
         /**
          * A client that has CLIENT_DENY_BLOCKING flag on
          * expects a reply per command and so can't execute MONITOR. */
-        addReplyError(c, "MONITOR is not allowed for DENY BLOCKING client");
+        addReplyError(c, "MONITOR isn't allowed for DENY BLOCKING client");
         return;
     }
 

src/server.h

@@ -773,6 +773,8 @@ typedef struct readyList {
                                            no AUTH is needed, and every
                                            connection is immediately
                                            authenticated. */
+#define USER_FLAG_ALLCHANNELS (1<<5)    /* The user can mention any Pub/Sub
+                                           channel. */
 typedef struct {
     sds name;       /* The username as an SDS string. */
     uint64_t flags; /* See USER_FLAG_* */
@@ -796,6 +798,10 @@ typedef struct {
     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][UNSUSBSCRIBE] command, unless the flag
+                        ALLCHANNELS is set in the user. */
 } user;
 
 /* With multiplexing we need to take per-client state.
@@ -1488,11 +1494,12 @@ struct redisServer {
     long long latency_monitor_threshold;
     dict *latency_events;
     /* ACLs */
-    char *acl_filename;     /* ACL Users file. NULL if not configured. */
+    char *acl_filename;           /* ACL Users file. NULL if not configured. */
     unsigned long acllog_max_len; /* Maximum length of the ACL LOG list. */
-    sds requirepass;        /* Remember the cleartext password set with the
+    sds requirepass;              /* Remember the cleartext password set with
-                               old "requirepass" directive for backward
+                                     the old "requirepass" directive for
-                               compatibility with Redis <= 5. */
+                                     backward compatibility with Redis <= 5. */
+    int acl_pubusub_default;      /* Default ACL pub/sub channels flag */
     /* Assert & bug reporting */
     int watchdog_period;  /* Software watchdog period in ms. 0 = off */
     /* System hardware info */
@@ -1961,17 +1968,19 @@ void sendChildCOWInfo(int ptype, char *pname);
 extern rax *Users;
 extern user *DefaultUser;
 void ACLInit(void);
-/* Return values for ACLCheckCommandPerm(). */
+/* Return values for ACLCheckCommandPerm() and ACLCheckPubsubPerm(). */
 #define ACL_OK 0
 #define ACL_DENIED_CMD 1
 #define ACL_DENIED_KEY 2
 #define ACL_DENIED_AUTH 3 /* Only used for ACL LOG entries. */
+#define ACL_DENIED_CHANNEL 4 /* Only used for pub/sub commands */
 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 ACLCheckCommandPerm(client *c, int *keyidxptr);
+int ACLCheckPubsubPerm(client *c, int idx, int count, int literal, int *idxptr);
 int ACLSetUser(user *u, const char *op, ssize_t oplen);
 sds ACLDefaultUserFirstPassword(void);
 uint64_t ACLGetCommandCategoryFlagByName(const char *name);
@@ -2096,6 +2105,7 @@ struct redisMemOverhead *getMemoryOverheadData(void);
 void freeMemoryOverheadData(struct redisMemOverhead *mh);
 void checkChildrenDone(void);
 int setOOMScoreAdj(int process_class);
+void rejectCommandFormat(client *c, const char *fmt, ...);
 
 #define RESTART_SERVER_NONE 0
 #define RESTART_SERVER_GRACEFULLY (1<<0)     /* Do proper shutdown. */

tests/unit/acl.tcl

@@ -81,6 +81,100 @@ start_server {tags {"acl"}} {
         set e
     } {*NOPERM*key*}
 
+    test {By default users are able to publish to any channel} {
+        r ACL setuser psuser on >pspass +acl +client +@pubsub
+        r AUTH psuser pspass
+        r PUBLISH foo bar
+    } {0}
+
+    test {By default users are able to subscribe to any channel} {
+        set rd [redis_deferring_client]
+        $rd AUTH psuser pspass
+        $rd read
+        $rd SUBSCRIBE foo
+        assert_match {subscribe foo 1} [$rd read]
+        $rd close
+    } {0}
+
+    test {By default users are able to subscribe to any pattern} {
+        set rd [redis_deferring_client]
+        $rd AUTH psuser pspass
+        $rd read
+        $rd PSUBSCRIBE bar*
+        assert_match {psubscribe bar\* 1} [$rd read]
+        $rd close
+    } {0}
+
+    test {It's possible to allow publishing to a subset of channels} {
+        r ACL setuser psuser resetchannels &foo:1 &bar:*
+        assert_equal {0} [r PUBLISH foo:1 somemessage]
+        assert_equal {0} [r PUBLISH bar:2 anothermessage]
+        catch {r PUBLISH zap:3 nosuchmessage} e
+        set e
+    } {*NOPERM*channel*}
+
+    test {It's possible to allow subscribing to a subset of channels} {
+        set rd [redis_deferring_client]
+        $rd AUTH psuser pspass
+        $rd read
+        $rd SUBSCRIBE foo:1
+        assert_match {subscribe foo:1 1} [$rd read]
+        $rd SUBSCRIBE bar:2
+        assert_match {subscribe bar:2 2} [$rd read]
+        $rd SUBSCRIBE zap:3
+        catch {$rd read} e
+        set e
+    } {*NOPERM*channel*}
+
+    test {It's possible to allow subscribing to a subset of channel patterns} {
+        set rd [redis_deferring_client]
+        $rd AUTH psuser pspass
+        $rd read
+        $rd PSUBSCRIBE foo:1
+        assert_match {psubscribe foo:1 1} [$rd read]
+        $rd PSUBSCRIBE bar:*
+        assert_match {psubscribe bar:\* 2} [$rd read]
+        $rd PSUBSCRIBE bar:baz
+        catch {$rd read} e
+        set e
+    } {*NOPERM*channel*}
+    
+    test {Subscribers are killed when revoked of channel permission} {
+        set rd [redis_deferring_client]
+        r ACL setuser psuser resetchannels &foo:1
+        $rd AUTH psuser pspass
+        $rd CLIENT SETNAME deathrow
+        $rd SUBSCRIBE foo:1
+        r ACL setuser psuser resetchannels
+        assert_no_match {*deathrow*} [r CLIENT LIST]
+        $rd close
+    } {0}
+
+    test {Subscribers are killed when revoked of pattern permission} {
+        set rd [redis_deferring_client]
+        r ACL setuser psuser resetchannels &bar:*
+        $rd AUTH psuser pspass
+        $rd CLIENT SETNAME deathrow
+        $rd PSUBSCRIBE bar:*
+        r ACL setuser psuser resetchannels
+        assert_no_match {*deathrow*} [r CLIENT LIST]
+        $rd close
+    } {0}
+
+    test {Subscribers are pardoned if literal permissions are retained and/or gaining allchannels} {
+        set rd [redis_deferring_client]
+        r ACL setuser psuser resetchannels &foo:1 &bar:*
+        $rd AUTH psuser pspass
+        $rd CLIENT SETNAME pardoned
+        $rd SUBSCRIBE foo:1
+        $rd PSUBSCRIBE bar:*
+        r ACL setuser psuser resetchannels &foo:1 &bar:* &baz:qaz &zoo:*
+        assert_match {*pardoned*} [r CLIENT LIST]
+        r ACL setuser psuser allchannels
+        assert_match {*pardoned*} [r CLIENT LIST]
+        $rd close
+    } {0}
+
     test {Users can be configured to authenticate with any password} {
         r ACL setuser newuser nopass
         r AUTH newuser zipzapblabla
@@ -166,6 +260,7 @@ start_server {tags {"acl"}} {
         r ACL LOG RESET
         r ACL setuser antirez >foo on +set ~object:1234
         r ACL setuser antirez +eval +multi +exec
+        r ACL setuser antirez resetchannels +publish
         r AUTH antirez foo
         catch {r GET foo}
         r AUTH default ""
@@ -195,6 +290,15 @@ start_server {tags {"acl"}} {
         assert {[dict get $entry object] eq {somekeynotallowed}}
     }
 
+    test {ACL LOG is able to log channel access violations and channel name} {
+        r AUTH antirez foo
+        catch {r PUBLISH somechannelnotallowed nullmsg}
+        r AUTH default ""
+        set entry [lindex [r ACL LOG] 0]
+        assert {[dict get $entry reason] eq {channel}}
+        assert {[dict get $entry object] eq {somechannelnotallowed}}
+    }
+
     test {ACL LOG RESET is able to flush the entries in the log} {
         r ACL LOG RESET
         assert {[llength [r ACL LOG]] == 0}

tests/unit/moduleapi/blockedclient.tcl

@@ -63,7 +63,7 @@ start_server {tags {"modules"}} {
             r do_rm_call monitor
         } e
         set e
-    } {*MONITOR is not allow*}
+    } {*ERR*DENY BLOCKING*}
 
     test {subscribe disallow inside RM_Call} {
         set e {}
@@ -71,5 +71,5 @@ start_server {tags {"modules"}} {
             r do_rm_call subscribe x
         } e
         set e
-    } {*subscribe is not allow*}
+    } {*ERR*DENY BLOCKING*}
 }