Modules API: Add RedisModule_ACLCheckKeyPrefixPermissions (#13666)

This PR introduces a new API function to the Redis Module API:
```
int RedisModule_ACLCheckKeyPrefixPermissions(RedisModuleUser *user, RedisModuleString *prefix, int flags);
```
Purpose:
The function checks if a given user has access permissions to any key
that match a specific prefix. This validation is based on the user’s ACL
permissions and the specified flags.

Note, this prefix-based approach API may fail to detect prefixes that
are individually uncovered but collectively covered by the patterns. For
example the prefix `ID-*` is not fully included in pattern `ID-[0]*` and
is not fully included in pattern `ID-[^0]*` but it is fully included in
the set of patterns `{ID-[0]*, ID-[^0]*}`

src/acl.c

@@ -1577,14 +1577,22 @@ static int ACLSelectorCheckKey(aclSelector *selector, const char *key, int keyle
     if (keyspec_flags & CMD_KEY_DELETE) key_flags |= ACL_WRITE_PERMISSION;
     if (keyspec_flags & CMD_KEY_UPDATE) key_flags |= ACL_WRITE_PERMISSION;
 
+    /* Is given key represent a prefix of a set of keys */
+    int prefix = keyspec_flags & CMD_KEY_PREFIX;
+
     /* 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;
+        if (prefix) {
+            if (prefixmatch(pattern->pattern,plen,key,keylen,0))
+                return ACL_OK;
+        } else {
+            if (stringmatchlen(pattern->pattern, plen, key, keylen, 0))
+                return ACL_OK;
+        }
     }
     return ACL_DENIED_KEY;
 }

src/module.c

@@ -9774,6 +9774,45 @@ int RM_ACLCheckKeyPermissions(RedisModuleUser *user, RedisModuleString *key, int
     return REDISMODULE_OK;
 }
 
+/* Check if the user can access keys matching the given key prefix according to the ACLs 
+ * attached to the user and the flags representing key access. The flags are the same that 
+ * are used in the keyspec for logical operations. These flags are documented in 
+ * RedisModule_SetCommandInfo as the REDISMODULE_CMD_KEY_ACCESS, 
+ * REDISMODULE_CMD_KEY_UPDATE, REDISMODULE_CMD_KEY_INSERT, and REDISMODULE_CMD_KEY_DELETE flags.
+ * 
+ * If no flags are supplied, the user is still required to have some access to keys matching 
+ * the prefix for this command to return successfully.
+ *
+ * If the user is able to access keys matching the prefix, then REDISMODULE_OK is returned.
+ * Otherwise, REDISMODULE_ERR is returned and errno is set to one of the following values:
+ * 
+ * * EINVAL: The provided flags are invalid.
+ * * EACCES: The user does not have permission to access keys matching the prefix.
+ */
+int RM_ACLCheckKeyPrefixPermissions(RedisModuleUser *user, RedisModuleString *prefix, int flags) {
+    const int allow_mask = (REDISMODULE_CMD_KEY_ACCESS
+                            | REDISMODULE_CMD_KEY_INSERT
+                            | REDISMODULE_CMD_KEY_DELETE
+                            | REDISMODULE_CMD_KEY_UPDATE);
+
+    if ((flags & allow_mask) != flags) {
+        errno = EINVAL;
+        return REDISMODULE_ERR;
+    }
+
+    int keyspec_flags = moduleConvertKeySpecsFlags(flags, 0);
+
+    /* Add the prefix flag to the keyspec flags */
+    keyspec_flags |= CMD_KEY_PREFIX;
+
+    if (ACLUserCheckKeyPerm(user->user, prefix->ptr, sdslen(prefix->ptr), keyspec_flags) != ACL_OK) {
+        errno = EACCES;
+        return REDISMODULE_ERR;
+    }
+
+    return REDISMODULE_OK;
+}
+
 /* Check if the pubsub channel can be accessed by the user based off of the given
  * access flags. See RM_ChannelAtPosWithFlags for more information about the
  * possible flags that can be passed in.
@@ -14186,6 +14225,7 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(GetModuleUserFromUserName);
     REGISTER_API(ACLCheckCommandPermissions);
     REGISTER_API(ACLCheckKeyPermissions);
+    REGISTER_API(ACLCheckKeyPrefixPermissions);
     REGISTER_API(ACLCheckChannelPermissions);
     REGISTER_API(ACLAddLogEntry);
     REGISTER_API(ACLAddLogEntryByUserName);

src/redismodule.h

@@ -1288,6 +1288,7 @@ REDISMODULE_API RedisModuleString * (*RedisModule_GetCurrentUserName)(RedisModul
 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, int flags) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_ACLCheckKeyPrefixPermissions)(RedisModuleUser *user, RedisModuleString *prefix, 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, RedisModuleACLLogEntryReason reason) REDISMODULE_ATTR;
 REDISMODULE_API void (*RedisModule_ACLAddLogEntryByUserName)(RedisModuleCtx *ctx, RedisModuleString *user, RedisModuleString *object, RedisModuleACLLogEntryReason reason) REDISMODULE_ATTR;
@@ -1657,6 +1658,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(GetModuleUserFromUserName);
     REDISMODULE_GET_API(ACLCheckCommandPermissions);
     REDISMODULE_GET_API(ACLCheckKeyPermissions);
+    REDISMODULE_GET_API(ACLCheckKeyPrefixPermissions);    
     REDISMODULE_GET_API(ACLCheckChannelPermissions);
     REDISMODULE_GET_API(ACLAddLogEntry);
     REDISMODULE_GET_API(ACLAddLogEntryByUserName);

src/server.h

@@ -274,6 +274,7 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
                                        * out to all keys it should cover */
 #define CMD_KEY_VARIABLE_FLAGS (1ULL<<10)  /* Means that some keys might have
                                             * different flags depending on arguments */
+#define CMD_KEY_PREFIX (1ULL<<11) /* Given key represents a prefix of a set of keys */
 
 /* Key flags for when access type is unknown */
 #define CMD_KEY_FULL_ACCESS (CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_UPDATE)

src/util.c

@@ -198,6 +198,43 @@ static int stringmatchlen_impl(const char *pattern, int patternLen,
     return 0;
 }
 
+/* 
+ * glob-style pattern matching to check if a given pattern fully includes 
+ * the prefix of a string. For the match to succeed, the pattern must end with 
+ * an unescaped '*' character.
+ * 
+ * Returns: 1 if the `pattern` fully matches the `prefixStr`. Returns 0 otherwise.
+ */
+int prefixmatch(const char *pattern, int patternLen,
+                const char *prefixStr, int prefixStrLen, int nocase) {
+    int skipLongerMatches = 0;
+    
+    /* Step 1: Verify if the pattern matches the prefix string completely. */
+    if (!stringmatchlen_impl(pattern, patternLen, prefixStr, prefixStrLen, nocase, &skipLongerMatches, 0))
+        return 0;
+
+    /* Step 2: Verify that the pattern ends with an unescaped '*', indicating
+     * it can match any suffix of the string beyond the prefix. This check
+     * remains outside stringmatchlen_impl() to keep its complexity manageable.
+     */
+    if (pattern[patternLen - 1] != '*' || patternLen == 0)
+        return 0;
+
+    /* Count backward the number of consecutive backslashes preceding the '*'
+     * to determine if the '*' is escaped. */
+    int backslashCount = 0;
+    for (int i = patternLen - 2; i >= 0; i--) {
+        if (pattern[i] == '\\')
+            ++backslashCount;
+        else
+            break; /* Stop counting when a non-backslash character is found. */
+    }
+
+    /* Return 1 if the '*' is not escaped (i.e., even count), 0 otherwise. */
+    return (backslashCount % 2 == 0);
+}
+
+/* Glob-style pattern matching to a string. */
 int stringmatchlen(const char *pattern, int patternLen,
         const char *string, int stringLen, int nocase) {
     int skipLongerMatches = 0;

src/util.h

@@ -36,6 +36,8 @@ typedef enum {
     LD_STR_HEX       /* %La */
 } ld2string_mode;
 
+int prefixmatch(const char *pattern, int patternLen, const char *prefixStr, 
+                int prefixStrLen, int nocase);
 int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase);
 int stringmatch(const char *p, const char *s, int nocase);
 int stringmatchlen_fuzz_test(void);

tests/modules/aclcheck.c

@@ -51,6 +51,52 @@ int set_aclcheck_key(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     return REDISMODULE_OK;
 }
 
+/* A wrap for SET command with ACL check on the key. */
+int set_aclcheck_prefixkey(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (argc < 4) {
+        return RedisModule_WrongArity(ctx);
+    }
+
+    int permissions;
+    const char *flags = RedisModule_StringPtrLen(argv[1], NULL);
+
+    if (!strcasecmp(flags, "W")) {
+        permissions = REDISMODULE_CMD_KEY_UPDATE;
+    } else if (!strcasecmp(flags, "R")) {
+        permissions = REDISMODULE_CMD_KEY_ACCESS;
+    } else if (!strcasecmp(flags, "*")) {
+        permissions = REDISMODULE_CMD_KEY_UPDATE | REDISMODULE_CMD_KEY_ACCESS;
+    } else if (!strcasecmp(flags, "~")) {
+        permissions = 0; /* Requires either read or write */
+    } 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_ACLCheckKeyPrefixPermissions(user, argv[2], permissions);
+    if (ret != 0) {
+        RedisModule_ReplyWithError(ctx, "DENIED KEY");
+        RedisModule_FreeModuleUser(user);
+        RedisModule_FreeString(ctx, user_name);
+        return REDISMODULE_OK;
+    }
+
+    RedisModuleCallReply *rep = RedisModule_Call(ctx, "SET", "v", argv + 3, argc - 3);
+    if (!rep) {
+        RedisModule_ReplyWithError(ctx, "NULL reply returned");
+    } else {
+        RedisModule_ReplyWithCallReply(ctx, rep);
+        RedisModule_FreeCallReply(rep);
+    }
+
+    RedisModule_FreeModuleUser(user);
+    RedisModule_FreeString(ctx, user_name);
+    return REDISMODULE_OK;
+}
+
 /* A wrap for PUBLISH command with ACL check on the channel. */
 int publish_aclcheck_channel(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     if (argc != 3) {
@@ -247,6 +293,9 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
     if (RedisModule_CreateCommand(ctx,"aclcheck.set.check.key", set_aclcheck_key,"write",0,0,0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx,"aclcheck.set.check.prefixkey", set_aclcheck_prefixkey,"write",0,0,0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;    
+
     if (RedisModule_CreateCommand(ctx,"block.commands.outside.onload", commandBlockCheck,"write",0,0,0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 

tests/unit/moduleapi/aclcheck.tcl

@@ -18,10 +18,43 @@ start_server {tags {"modules acl"}} {
         assert {[dict get $entry object] eq {set}}
         assert {[dict get $entry reason] eq {command}}
     }
+        
+    test {test module check acl for key prefix permission} {
+        r acl setuser default +set resetkeys ~CART* %W~ORDER* %R~PRODUCT* ~ESCAPED_STAR\\* ~NON_ESCAPED_STAR\\\\*
+        
+        # check for key permission of prefix CART* (READ+WRITE)
+        catch {r aclcheck.set.check.prefixkey "~" CAR CART_CLOTHES_7 5} e
+        assert_match "*DENIED KEY*" $e
+        assert_equal [r aclcheck.set.check.prefixkey "~" CART CART 5] OK
+        assert_equal [r aclcheck.set.check.prefixkey "W" CART_BOOKS CART_BOOKS_12 5] OK
+        assert_equal [r aclcheck.set.check.prefixkey "R" CART_CLOTHES CART_CLOTHES_7 5] OK
+        
+        # check for key permission of prefix ORDER* (WRITE)
+        catch {r aclcheck.set.check.prefixkey "~" ORDE ORDER_2024_155351 5} e
+        assert_match "*DENIED KEY*" $e
+        assert_equal [r aclcheck.set.check.prefixkey "~" ORDER ORDER 5] OK
+        assert_equal [r aclcheck.set.check.prefixkey "W" ORDER_2024 ORDER_2024_564879 5] OK
+        assert_equal [r aclcheck.set.check.prefixkey "~" ORDER_2023 ORDER_2023_564879 5] OK
+        catch {r aclcheck.set.check.prefixkey "R" ORDER_2023 ORDER_2023_564879 5}
+        assert_match "*DENIED KEY*" $e
+        
+        # check for key permission of prefix PRODUCT* (READ)
+        catch {r aclcheck.set.check.prefixkey "~" PRODUC PRODUCT_CLOTHES_753376 5} e
+        assert_match "*DENIED KEY*" $e
+        assert_equal [r aclcheck.set.check.prefixkey "~" PRODUCT PRODUCT 5] OK
+        assert_equal [r aclcheck.set.check.prefixkey "~" PRODUCT_BOOKS PRODUCT_BOOKS_753376 5] OK
+        
+        # pattern ends with a escaped '*' character should not be counted as a prefix
+        catch {r aclcheck.set.check.prefixkey "~" ESCAPED_STAR ESCAPED_STAR_12 5} e
+        assert_match "*DENIED KEY*" $e
+        catch {r aclcheck.set.check.prefixkey "~" ESCAPED_STAR* ESCAPED_STAR* 5} e
+        assert_match "*DENIED KEY*" $e        
+        assert_equal [r aclcheck.set.check.prefixkey "~" NON_ESCAPED_STAR\\ NON_ESCAPED_STAR\\clothes 5] OK
+    }    
 
     test {test module check acl for key perm} {
         # give permission for SET and block all keys but x(READ+WRITE), y(WRITE), z(READ)
-        r acl setuser default +set resetkeys ~x %W~y %R~z
+        r acl setuser default +set resetkeys ~x %W~y %R~z ~ESCAPED_STAR\\*
 
         assert_equal [r aclcheck.set.check.key "*" x 5] OK
         catch {r aclcheck.set.check.key "*" v 5} e
@@ -40,6 +73,9 @@ start_server {tags {"modules acl"}} {
         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
+        
+        # check pattern ends with escaped '*' character
+        assert_equal [r aclcheck.set.check.key "~" ESCAPED_STAR* 5] OK
     }
 
     test {test module check acl for module user} {