wip

fakeredis/_basefakesocket.py

@@ -190,7 +190,7 @@ def _process_command(self, fields: List[bytes]) -> None:
         cmd, cmd_arguments = _extract_command(fields)
         try:
             func, sig = self._name_to_func(cmd)
-            self._server.acl.validate_command(self._current_user, fields)  # ACL check
+            self._server.acl.validate_command(self._current_user, self._client_info, fields)  # ACL check
             with self._server.lock:
                 # Clean out old connections
                 while True:

fakeredis/_msgs.py

@@ -124,6 +124,7 @@
 MISSING_ACLFILE_CONFIG = "ERR This Redis instance is not configured to use an ACL file. You may want to specify users via the ACL SETUSER command and then issue a CONFIG REWRITE (assuming you have a Redis configuration file set) in order to store users in the Redis configuration."
 
 NO_PERMISSION_ERROR = "NOPERM User {} has no permissions to run the '{}' command"
+NO_PERMISSION_KEY_ERROR = "NOPERM No permissions to access a key"
 
 # Command flags
 FLAG_NO_SCRIPT = "s"  # Command not allowed in scripts

fakeredis/model/_acl.py

@@ -1,5 +1,6 @@
+import fnmatch
 import hashlib
-from typing import Dict, Set, List, Union, Optional
+from typing import Dict, Set, List, Union, Optional, Sequence
 
 from fakeredis import _msgs as msgs
 from ._command_info import get_commands_by_category, get_command_info
@@ -62,17 +63,43 @@ 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)
+    @staticmethod
+    def _get_command_info(fields: List[bytes]):
+        command = fields[0].lower()
         command_info = get_command_info(command)
+        if not command_info and len(fields) > 1:
+            command = command + b" " + fields[1].lower()
+            command_info = get_command_info(command)
+        return command_info
+
+    def command_allowed(self, fields: List[bytes]) -> bool:
+        res = fields[0].lower() == b"auth" or self._commands.get(fields[0].lower(), False)
+        res = res or self._commands.get(b"@all", False)
+        command_info = self._get_command_info(fields)
         if not command_info:
             return res
         for category in command_info[6]:
             res = res or self._commands.get(category, False)
         return res
 
+    def _get_keys(self, fields: List[bytes]) -> List[bytes]:
+        command_info = self._get_command_info(fields)
+        if not command_info:
+            return []
+        first_key, last_key, step = command_info[3:6]
+        if first_key == 0:
+            return []
+        last_key = (last_key + 1) if last_key >= 0 else last_key
+        step = step + 1
+        return fields[first_key : last_key + 1 : step]
+
+    def keys_not_allowed(self, fields: List[bytes]) -> List[bytes]:
+        keys = self._get_keys(fields)
+        res = set()
+        for pat in self._key_patterns:
+            res = res.union(fnmatch.filter(keys, pat))
+        return list(set(keys) - res)
+
     def set_nopass(self) -> None:
         self._nopass = True
         self._passwords.clear()
@@ -306,15 +333,18 @@ def add_log_record(
         )
         self._log.append(entry)
 
-    def validate_command(self, username: bytes, fields: List[bytes]):
+    def validate_command(self, username: bytes, client_info: bytes, fields: List[bytes]):
         if username not in self._user_acl:
             return
         user_acl = self._user_acl[username]
         if not user_acl.enabled:
             raise SimpleError("User disabled")
 
-        command, args = fields[0], fields[1:]
-
-        if not user_acl.command_allowed(command):
-            raise SimpleError(msgs.NO_PERMISSION_ERROR.format(username.decode(), command.lower().decode()))
+        if not user_acl.command_allowed(fields):
+            self.add_log_record(b"command", b"toplevel", fields[0], username, client_info)
+            raise SimpleError(msgs.NO_PERMISSION_ERROR.format(username.decode(), fields[0].lower().decode()))
+        keys_not_allowed = user_acl.keys_not_allowed(fields)
+        if len(keys_not_allowed) > 0:
+            self.add_log_record(b"key", b"toplevel", keys_not_allowed[0], username, client_info)
+            raise SimpleError(msgs.NO_PERMISSION_KEY_ERROR)
         # todo

scripts/generate_command_info.py

@@ -98,6 +98,14 @@ def get_command_info(cmd_name: str, all_commands: Dict[str, Any]) -> List[Any]:
             if cmd not in implemented:
                 continue
             command_info_dict[cmd] = get_command_info(cmd, cmds)
+            subcommand = cmd.split(" ")
+            if len(subcommand) > 1:
+                (
+                    command_info_dict.setdefault(subcommand[0], [subcommand[0], -1, [], 0, 0, 0, [], [], [], []])[
+                        9
+                    ].append(command_info_dict[cmd])
+                )
+
             print(command_info_dict[cmd])
     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

@@ -317,15 +317,17 @@ def teardown():
     assert r.get("cache:0") == b"1"
 
     # Invalid operation
-    with pytest.raises(exceptions.NoPermissionError) as command_not_permitted:
+    with pytest.raises(exceptions.NoPermissionError) as ctx:
         r.hset("cache:0", "hkey", "hval")
 
-    assert str(command_not_permitted.value) == "User fredis-py-user has no permissions to run the 'hset' command"
+    assert str(ctx.value) == "User fredis-py-user has no permissions to run the 'hset' command"
 
     # Invalid key
-    with pytest.raises(exceptions.NoPermissionError):
+    with pytest.raises(exceptions.NoPermissionError) as ctx:
         r.get("violated_cache:0")
 
+    assert str(ctx.value) == "No permissions to access a key"
+
     r.auth("", "default")
     log = r.acl_log()
     assert isinstance(log, list)