feat: add validation of acl users (#1743)

* add validation for categories
* add tests

src/core/overloaded.h

@@ -3,6 +3,8 @@
 //
 //
 
+#pragma once
+
 template <class... Ts> struct Overloaded : Ts... { using Ts::operator()...; };
 
 template <class... Ts> Overloaded(Ts...) -> Overloaded<Ts...>;

src/server/CMakeLists.txt

@@ -25,7 +25,8 @@ add_library(dragonfly_lib channel_store.cc command_registry.cc
             zset_family.cc version.cc bitops_family.cc container_utils.cc io_utils.cc
             serializer_commons.cc journal/serializer.cc journal/executor.cc journal/streamer.cc
             top_keys.cc multi_command_squasher.cc hll_family.cc cluster/cluster_config.cc
-            cluster/cluster_family.cc acl/user.cc acl/user_registry.cc acl/acl_family.cc)
+            cluster/cluster_family.cc acl/user.cc acl/user_registry.cc acl/acl_family.cc
+            acl/validator.cc)
 
 
 cxx_link(dragonfly_lib dfly_transaction dfly_facade redis_lib aws_lib strings_lib html_lib

src/server/acl/user.h

@@ -16,10 +16,6 @@
 
 namespace dfly::acl {
 
-// TODO implement these
-//#bool CheckIfCommandAllowed(uint64_t command_id, const CommandId& command);
-//#bool CheckIfAclCategoryAllowed(uint64_t command_id, const CommandId& command);
-
 class User final {
  public:
   struct UpdateRequest {

src/server/acl/validator.cc

@@ -0,0 +1,19 @@
+// Copyright 2023, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+
+#include "server/acl/validator.h"
+
+#include "server/server_state.h"
+
+namespace dfly::acl {
+
+[[nodiscard]] bool IsUserAllowedToInvokeCommand(const ConnectionContext& cntx,
+                                                const facade::CommandId& id) {
+  auto& registry = *ServerState::tlocal()->user_registry;
+  auto credentials = registry.GetCredentials(cntx.authed_username);
+  auto command_credentials = id.acl_categories();
+  return (credentials.acl_categories & command_credentials) != 0;
+}
+
+}  // namespace dfly::acl

src/server/acl/validator.h

@@ -0,0 +1,16 @@
+// Copyright 2023, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+
+#pragma once
+
+#include <string_view>
+
+#include "facade/command_id.h"
+#include "server/conn_context.h"
+
+namespace dfly::acl {
+
+bool IsUserAllowedToInvokeCommand(const ConnectionContext& cntx, const facade::CommandId& id);
+
+}

src/server/conn_context.h

@@ -196,7 +196,7 @@ class ConnectionContext : public facade::ConnectionContext {
   // Reference to a FlowInfo for this connection if from a master to a replica.
   FlowInfo* replication_flow;
 
-  std::optional<std::string> authed_username;
+  std::string authed_username{"default"};
 
  private:
   void EnableMonitoring(bool enable) {

src/server/main_service.cc

@@ -27,6 +27,7 @@ extern "C" {
 #include "facade/reply_capture.h"
 #include "server/acl/acl_commands_def.h"
 #include "server/acl/acl_family.h"
+#include "server/acl/validator.h"
 #include "server/bitops_family.h"
 #include "server/cluster/cluster_family.h"
 #include "server/conn_context.h"
@@ -784,11 +785,27 @@ OpStatus CheckKeysDeclared(const ConnectionState::ScriptInfo& eval_info, const C
   return OpStatus::OK;
 }
 
-optional<ErrorReply> Service::VerifyCommandExecution(const CommandId* cid) {
-  // TODO: Move OOM check here
+static optional<ErrorReply> VerifyConnectionAclStatus(const CommandId* cid,
+                                                      const ConnectionContext* cntx,
+                                                      string_view error_msg) {
+  // If we are on a squashed context we need to use the owner, because the
+  // context we are operating on is a stub and the acl username is not copied
+  // See: MultiCommandSquasher::SquashedHopCb
+  if (cntx->conn_state.squashing_info)
+    cntx = cntx->conn_state.squashing_info->owner;
+
+  if (!acl::IsUserAllowedToInvokeCommand(*cntx, *cid)) {
+    return ErrorReply(absl::StrCat("NOPERM: ", cntx->authed_username, " ", error_msg));
+  }
   return nullopt;
 }
 
+optional<ErrorReply> Service::VerifyCommandExecution(const CommandId* cid,
+                                                     const ConnectionContext* cntx) {
+  // TODO: Move OOM check here
+  return VerifyConnectionAclStatus(cid, cntx, "ACL rules changed between the MULTI and EXEC");
+}
+
 std::optional<ErrorReply> Service::VerifyCommandState(const CommandId* cid, CmdArgList tail_args,
                                                       const ConnectionContext& dfly_cntx) {
   DCHECK(cid);
@@ -866,7 +883,7 @@ std::optional<ErrorReply> Service::VerifyCommandState(const CommandId* cid, CmdA
       return ErrorReply{status};
   }
 
-  return nullopt;
+  return VerifyConnectionAclStatus(cid, &dfly_cntx, "has no ACL permissions");
 }
 
 void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx) {
@@ -980,7 +997,7 @@ bool Service::InvokeCmd(const CommandId* cid, CmdArgList tail_args, ConnectionCo
   DCHECK(cid);
   DCHECK(!cid->Validate(tail_args));
 
-  if (auto err = VerifyCommandExecution(cid); err) {
+  if (auto err = VerifyCommandExecution(cid, cntx); err) {
     (*cntx)->SendError(std::move(*err));
     return true;  // return false only for internal error aborts
   }

src/server/main_service.h

@@ -55,7 +55,8 @@ class Service : public facade::ServiceInterface {
 
   // Verify command can be executed now (check out of memory), always called immediately before
   // execution
-  std::optional<facade::ErrorReply> VerifyCommandExecution(const CommandId* cid);
+  std::optional<facade::ErrorReply> VerifyCommandExecution(const CommandId* cid,
+                                                           const ConnectionContext* cntx);
 
   // Verify command prepares excution in correct state.
   // It's usually called before command execution. Only for multi/exec transactions it's checked

tests/dragonfly/acl_family_test.py

@@ -59,21 +59,133 @@ async def test_acl_setuser(async_client):
 
 @pytest.mark.asyncio
 async def test_acl_auth(async_client):
-    await async_client.execute_command("ACL SETUSER kostas >mypass")
+    await async_client.execute_command("ACL SETUSER shahar >mypass")
 
     with pytest.raises(redis.exceptions.ResponseError):
-        await async_client.execute_command("AUTH kostas wrong_pass")
+        await async_client.execute_command("AUTH shahar wrong_pass")
 
     # This should fail because user is inactive
     with pytest.raises(redis.exceptions.ResponseError):
-        await async_client.execute_command("AUTH kostas mypass")
+        await async_client.execute_command("AUTH shahar mypass")
 
     # Activate user
-    await async_client.execute_command("ACL SETUSER kostas ON")
+    await async_client.execute_command("ACL SETUSER shahar ON +@fast")
 
-    result = await async_client.execute_command("AUTH kostas mypass")
-    result == "ok"
+    result = await async_client.execute_command("AUTH shahar mypass")
+    assert result == "OK"
 
     # Let's also try default
     result = await async_client.execute_command("AUTH default nopass")
-    result == "ok"
+    assert result == "OK"
+
+
+@pytest.mark.asyncio
+async def test_acl_categories(async_client):
+    await async_client.execute_command("ACL SETUSER vlad ON >mypass +@string +@list +@connection")
+
+    result = await async_client.execute_command("AUTH vlad mypass")
+    assert result == "OK"
+
+    result = await async_client.execute_command("SET foo bar")
+    assert result == "OK"
+
+    result = await async_client.execute_command("LPUSH mykey space_monkey")
+    assert result == 1
+
+    # This should fail, vlad does not have @admin
+    with pytest.raises(redis.exceptions.ResponseError):
+        await async_client.execute_command("ACL SETUSER vlad ON >mypass")
+
+    # This should fail, vlad does not have @sortedset
+    with pytest.raises(redis.exceptions.ResponseError):
+        await async_client.execute_command("ZADD myset 1 two")
+
+    result = await async_client.execute_command("AUTH default nopass")
+    assert result == "OK"
+
+    # Make vlad an admin
+    await async_client.execute_command("ACL SETUSER vlad -@string")
+    assert result == "OK"
+
+    result = await async_client.execute_command("AUTH vlad mypass")
+    assert result == "OK"
+
+    with pytest.raises(redis.exceptions.ResponseError):
+        await async_client.execute_command("GET foo")
+
+    result = await async_client.execute_command("AUTH default nopass")
+    assert result == "OK"
+
+    # Vlad goes rogue starts giving admin stats to random users
+    await async_client.execute_command("ACL SETUSER adi >adi +@admin")
+    assert result == "OK"
+
+    # Vlad can now execute everything
+    await async_client.execute_command("ACL SETUSER vlad +@all")
+    assert result == "OK"
+
+    await async_client.execute_command("ZADD myset 1 two")
+    assert result == "OK"
+
+
+@pytest.mark.asyncio
+async def test_acl_categories_multi_exec_squash(df_local_factory):
+    df = df_local_factory.create(multi_exec_squash=True, port=1111)
+
+    df.start()
+
+    client = aioredis.Redis(port=df.port)
+    res = await client.execute_command("ACL SETUSER kk ON >kk +@transaction +@string")
+    assert res == b"OK"
+
+    res = await client.execute_command("AUTH kk kk")
+    assert res == b"OK"
+
+    await client.execute_command("MULTI")
+    assert res == b"OK"
+    for x in range(33):
+        await client.execute_command(f"SET x{x} {x}")
+    await client.execute_command("EXEC")
+
+    client = aioredis.Redis(port=df.port)
+    await client.close()
+
+    # NOPERM while executing multi
+    await client.execute_command("ACL SETUSER kk -@string")
+    assert res == b"OK"
+    await client.execute_command("AUTH kk kk")
+    assert res == b"OK"
+    await client.execute_command("MULTI")
+    assert res == b"OK"
+
+    with pytest.raises(redis.exceptions.ResponseError):
+        await client.execute_command(f"SET x{x} {x}")
+    await client.close()
+
+    # NOPERM between multi and exec
+    admin_client = aioredis.Redis(port=df.port)
+    res = await client.execute_command("ACL SETUSER kk +@string")
+    assert res == b"OK"
+
+    client = aioredis.Redis(port=df.port)
+    res = await client.execute_command("AUTH kk kk")
+    assert res == b"OK"
+    # CLIENT has permissions, starts MULTI and issues a bunch of SET commands
+    await client.execute_command("MULTI")
+    assert res == b"OK"
+    for x in range(33):
+        await client.execute_command(f"SET x{x} {x}")
+
+    # ADMIN revokes permissions
+    res = await admin_client.execute_command("ACL SETUSER kk -@string")
+    assert res == b"OK"
+
+    res = await client.execute_command("EXEC")
+    # TODO(we need to fix this, basiscally SQUASHED/MULTI transaction commands
+    # return multiple errors for each command failed. Since the nature of the error
+    # is the same, that a rule has changed we should squash those error messages into
+    # one.
+    assert res[0].args[0] == "NOPERM: kk ACL rules changed between the MULTI and EXEC"
+
+    await admin_client.close()
+    await client.close()