sub-command support for ACL CAT and COMMAND LIST. redisCommand always stores fullname (#10127)

Summary of changes:
1. Rename `redisCommand->name` to `redisCommand->declared_name`, it is a
  const char * for native commands and SDS for module commands.
2. Store the [sub]command fullname in `redisCommand->fullname` (sds).
3. List subcommands in `ACL CAT`
4. List subcommands in `COMMAND LIST`
5. `moduleUnregisterCommands` now will also free the module subcommands.
6. RM_GetCurrentCommandName returns full command name

Other changes:
1. Add `addReplyErrorArity` and `addReplyErrorExpireTime`
2. Remove `getFullCommandName` function that now is useless.
3. Some cleanups about `fullname` since now it is SDS.
4. Delete `populateSingleCommand` function from server.h that is useless.
5. Added tests to cover this change.
6. Add some module unload tests and fix the leaks
7. Make error messages uniform, make sure they always contain the full command
  name and that it's quoted.
7. Fixes some typos

see the history in #9504, fixes #10124

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

1 files changed, 35 insertions, 26 deletions

diff --git a/src/acl.c b/src/acl.c
index 24a9e35b4..174861f4a 100644
--- a/src/acl.c
+++ b/src/acl.c
@@ -644,9 +644,7 @@ sds ACLDescribeSelectorCommandRulesSingleCommands(aclSelector *selector, aclSele
         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 = sdscatsds(rules,cmd->fullname);
             rules = sdscatlen(rules," ",1);
             ACLChangeSelectorPerm(fake_selector,cmd,userbit);
         }
@@ -660,9 +658,7 @@ sds ACLDescribeSelectorCommandRulesSingleCommands(aclSelector *selector, aclSele
         {
             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 = sdscatsds(rules,cmd->fullname);
                 rules = sdscatlen(rules,"|",1);
                 rules = sdscatsds(rules,selector->allowed_firstargs[cmd->id][j]);
                 rules = sdscatlen(rules," ",1);
@@ -994,10 +990,10 @@ cleanup:
  *              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 
+ * %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. 
+ *              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
@@ -1005,7 +1001,7 @@ cleanup:
  *              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.
+ * resetchannels            Flush the list of allowed channel patterns.
  */
 int ACLSetSelector(aclSelector *selector, const char* op, size_t oplen) {
     if (!strcasecmp(op,"allkeys") ||
@@ -1456,9 +1452,12 @@ int ACLAuthenticateUser(client *c, robj *username, robj *password) {
  * should have an assigned ID (that is used to index the bitmap). This function
  * creates such an ID: it uses sequential IDs, reusing the same ID for the same
  * command name, so that a command retains the same ID in case of modules that
- * are unloaded and later reloaded. */
-unsigned long ACLGetCommandID(const char *cmdname) {
-    sds lowername = sdsnew(cmdname);
+ * are unloaded and later reloaded.
+ *
+ * The function does not take ownership of the 'cmdname' SDS string.
+ * */
+unsigned long ACLGetCommandID(sds cmdname) {
+    sds lowername = sdsdup(cmdname);
     sdstolower(lowername);
     if (commandId == NULL) commandId = raxNew();
     void *id = raxFind(commandId,(unsigned char*)lowername,sdslen(lowername));
@@ -2374,7 +2373,7 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username
         le->object = object;
     } else {
         switch(reason) {
-            case ACL_DENIED_CMD: le->object = getFullCommandName(c->cmd); break;
+            case ACL_DENIED_CMD: le->object = sdsdup(c->cmd->fullname); break;
             case ACL_DENIED_KEY: le->object = sdsdup(c->argv[argpos]->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;
@@ -2435,6 +2434,26 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username
  * ACL related commands
  * ==========================================================================*/
 
+/* ACL CAT category */
+void aclCatWithFlags(client *c, dict *commands, uint64_t cflag, int *arraylen) {
+    dictEntry *de;
+    dictIterator *di = dictGetIterator(commands);
+
+    while ((de = dictNext(di)) != NULL) {
+        struct redisCommand *cmd = dictGetVal(de);
+        if (cmd->flags & CMD_MODULE) continue;
+        if (cmd->acl_categories & cflag) {
+            addReplyBulkCBuffer(c, cmd->fullname, sdslen(cmd->fullname));
+            (*arraylen)++;
+        }
+
+        if (cmd->subcommands_dict) {
+            aclCatWithFlags(c, cmd->subcommands_dict, cflag, arraylen);
+        }
+    }
+    dictReleaseIterator(di);
+}
+
 /* Add the formatted response from a single selector to the ACL GETUSER
  * response. This function returns the number of fields added. 
  * 
@@ -2686,22 +2705,12 @@ setuser_cleanup:
     } else if (!strcasecmp(sub,"cat") && c->argc == 3) {
         uint64_t cflag = ACLGetCommandCategoryFlagByName(c->argv[2]->ptr);
         if (cflag == 0) {
-            addReplyErrorFormat(c, "Unknown category '%s'", (char*)c->argv[2]->ptr);
+            addReplyErrorFormat(c, "Unknown category '%.128s'", (char*)c->argv[2]->ptr);
             return;
         }
         int arraylen = 0;
         void *dl = addReplyDeferredLen(c);
-        dictIterator *di = dictGetIterator(server.orig_commands);
-        dictEntry *de;
-        while ((de = dictNext(di)) != NULL) {
-            struct redisCommand *cmd = dictGetVal(de);
-            if (cmd->flags & CMD_MODULE) continue;
-            if (cmd->acl_categories & cflag) {
-                addReplyBulkCString(c,cmd->name);
-                arraylen++;
-            }
-        }
-        dictReleaseIterator(di);
+        aclCatWithFlags(c, server.orig_commands, cflag, &arraylen);
         setDeferredArrayLen(c,dl,arraylen);
     } else if (!strcasecmp(sub,"genpass") && (c->argc == 2 || c->argc == 3)) {
         #define GENPASS_MAX_BITS 4096
@@ -2809,7 +2818,7 @@ setuser_cleanup:
             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);
+                    "the '%s' command", c->cmd->fullname);
             } else if (result == ACL_DENIED_KEY) {
                 err = sdscatfmt(err, "This user has no permissions to access "
                     "the '%s' key", c->argv[idx + 3]->ptr);