wip

fakeredis/commands_mixins/acl_mixin.py

@@ -107,6 +107,7 @@ def acl_cat(self, *category: bytes) -> List[bytes]:
             res = get_categories()
         else:
             res = get_commands_by_category(category[0])
+            res = [cmd.replace(b" ", b"|") for cmd in res]
         return res
 
     @command(name="ACL GENPASS", fixed=(), repeat=(bytes,))

fakeredis/model/_acl.py

@@ -2,7 +2,7 @@
 from typing import Dict, Set, List, Union, Optional
 
 from fakeredis import _msgs as msgs
-from ._command_info import get_commands_by_category
+from ._command_info import get_commands_by_category, get_command_info
 from .._helpers import SimpleError, current_time
 
 
@@ -62,6 +62,17 @@ def reset(self):
         self._channel_patterns.clear()
         self._selectors.clear()
 
+    def command_allowed(self, command: bytes) -> bool:
+        command = command.lower()
+        res = command == b"auth" or self._commands.get(command, False)
+        res = res or self._commands.get(b"@all", False)
+        command_info = get_command_info(command)
+        if not command_info:
+            return res
+        for category in command_info[6]:
+            res = res or self._commands.get(category, False)
+        return res
+
     def set_nopass(self) -> None:
         self._nopass = True
         self._passwords.clear()
@@ -303,8 +314,7 @@ def validate_command(self, username: bytes, fields: List[bytes]):
             raise SimpleError("User disabled")
 
         command, args = fields[0], fields[1:]
-        if command.lower() == b"auth":
-            return
-        if command.lower() not in user_acl._commands:
+
+        if not user_acl.command_allowed(command):
             raise SimpleError(msgs.NO_PERMISSION_ERROR.format(username.decode(), command.lower().decode()))
         # todo

scripts/generate_command_info.py

@@ -16,6 +16,7 @@
 }
 that is used for the `COMMAND` redis command.
 """
+
 import json
 import os
 from typing import Any, List, Dict
@@ -28,10 +29,10 @@
 
 def implemented_commands() -> set:
     res = set(SUPPORTED_COMMANDS.keys())
-    if 'json.type' not in res:
-        raise ValueError('Make sure jsonpath_ng is installed to get accurate documentation')
-    if 'eval' not in res:
-        raise ValueError('Make sure lupa is installed to get accurate documentation')
+    if "json.type" not in res:
+        raise ValueError("Make sure jsonpath_ng is installed to get accurate documentation")
+    if "eval" not in res:
+        raise ValueError("Make sure lupa is installed to get accurate documentation")
     return res
 
 
@@ -64,17 +65,17 @@ def get_command_info(cmd_name: str, all_commands: Dict[str, Any]) -> List[Any]:
      9 Key specifications (as of Redis 7.0)
     10 Subcommands (as of Redis 7.0)
     """
-    print(f'Command {cmd_name}')
+    print(f"Command {cmd_name}")
     cmd_info = all_commands[cmd_name]
-    first_key = dict_deep_get(cmd_info, 'key_specs', 0, 'begin_search', 'spec', 'index', default_value=0)
-    last_key = dict_deep_get(cmd_info, 'key_specs', -1, 'begin_search', 'spec', 'index', default_value=0)
-    step = dict_deep_get(cmd_info, 'key_specs', 0, 'find_keys', 'spec', 'keystep', default_value=0)
+    first_key = dict_deep_get(cmd_info, "key_specs", 0, "begin_search", "spec", "index", default_value=0)
+    last_key = dict_deep_get(cmd_info, "key_specs", -1, "begin_search", "spec", "index", default_value=0)
+    step = dict_deep_get(cmd_info, "key_specs", 0, "find_keys", "spec", "keystep", default_value=0)
     tips = []  # todo
-    subcommands = [get_command_info(cmd, all_commands)
-                   for cmd in all_commands
-                   if cmd_name != cmd and cmd.startswith(cmd_name)]  # todo
+    subcommands = [
+        get_command_info(cmd, all_commands) for cmd in all_commands if cmd_name != cmd and cmd.startswith(cmd_name)
+    ]  # todo
     res = [
-        cmd_name.lower().replace(' ', '|'),
+        cmd_name.lower().replace(" ", "|"),
         cmd_info.get("arity", -1),
         cmd_info.get("command_flags", []),
         first_key,
@@ -88,15 +89,15 @@ def get_command_info(cmd_name: str, all_commands: Dict[str, Any]) -> List[Any]:
     return res
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     implemented = implemented_commands()
     command_info_dict: Dict[str, List[Any]] = dict()
     for cmd_meta in METADATA:
         cmds = download_single_stack_commands(cmd_meta.local_filename, cmd_meta.url)
         for cmd in cmds:
-            if cmd not in implemented or ' ' in cmd:
+            if cmd not in implemented:
                 continue
             command_info_dict[cmd] = get_command_info(cmd, cmds)
             print(command_info_dict[cmd])
-    with open(os.path.join(os.path.dirname(__file__), '..', 'fakeredis', 'commands.json'), 'w') as f:
+    with open(os.path.join(os.path.dirname(__file__), "..", "fakeredis", "commands.json"), "w") as f:
         json.dump(command_info_dict, f)

test/test_mixins/test_acl_commands.py

@@ -19,8 +19,9 @@ def test_acl_cat(r: redis.Redis):
         if "hpersist" in commands:
             commands.remove("hpersist")
         assert len(commands) > 0
+        commands = {cmd.replace(" ", "|") for cmd in commands}
         diff = set(commands) - set(r.acl_cat(cat))
-        assert len(diff) == 0
+        assert len(diff) == 0, f"Commands not found in category {cat}: {diff}"
 
 
 def test_acl_genpass(r: redis.Redis):