feat(AclFamily): add acl deluser (#1773)

* add acl deluser command
* add tests

Author: kostasrim

src/facade/conn_context.h

@@ -64,6 +64,10 @@ class ConnectionContext {
     rbuilder_->SendSimpleString(str);
   }
 
+  void SendOk() {
+    rbuilder_->SendOk();
+  }
+
   // connection state / properties.
   bool conn_closing : 1;
   bool req_auth : 1;

src/server/acl/acl_family.cc

@@ -188,6 +188,33 @@ void AclFamily::SetUser(CmdArgList args, ConnectionContext* cntx) {
   std::visit(Overloaded{error_case, update_case}, std::move(req));
 }
 
+void AclFamily::EvictOpenConnectionsOnAllProactors(std::string_view user) {
+  auto close_cb = [user]([[maybe_unused]] size_t id, util::Connection* conn) {
+    DCHECK(conn);
+    auto connection = static_cast<facade::Connection*>(conn);
+    auto ctx = static_cast<ConnectionContext*>(connection->cntx());
+    if (ctx && ctx->authed_username == user) {
+      connection->ShutdownSelf();
+    }
+  };
+
+  if (main_listener_) {
+    main_listener_->TraverseConnections(close_cb);
+  }
+}
+
+void AclFamily::DelUser(CmdArgList args, ConnectionContext* cntx) {
+  std::string_view username = facade::ToSV(args[0]);
+  auto& registry = *ServerState::tlocal()->user_registry;
+  if (!registry.RemoveUser(username)) {
+    cntx->SendError(absl::StrCat("User ", username, " does not exist"));
+    return;
+  }
+
+  EvictOpenConnectionsOnAllProactors(username);
+  cntx->SendOk();
+}
+
 using CI = dfly::CommandId;
 
 using MemberFunc = void (AclFamily::*)(CmdArgList args, ConnectionContext* cntx);
@@ -201,6 +228,7 @@ inline CommandId::Handler HandlerFunc(AclFamily* acl, MemberFunc f) {
 constexpr uint32_t kAcl = acl::CONNECTION;
 constexpr uint32_t kList = acl::ADMIN | acl::SLOW | acl::DANGEROUS;
 constexpr uint32_t kSetUser = acl::ADMIN | acl::SLOW | acl::DANGEROUS;
+constexpr uint32_t kDelUser = acl::ADMIN | acl::SLOW | acl::DANGEROUS;
 
 // We can't implement the ACL commands and its respective subcommands LIST, CAT, etc
 // the usual way, (that is, one command called ACL which then dispatches to the subcommand
@@ -215,6 +243,8 @@ void AclFamily::Register(dfly::CommandRegistry* registry) {
       List);
   *registry << CI{"ACL SETUSER", CO::ADMIN | CO::NOSCRIPT | CO::LOADING, -2, 0, 0, 0, acl::kSetUser}
                    .HFUNC(SetUser);
+  *registry << CI{"ACL DELUSER", CO::ADMIN | CO::NOSCRIPT | CO::LOADING, 2, 0, 0, 0, acl::kDelUser}
+                   .HFUNC(DelUser);
 }
 
 #undef HFUNC

src/server/acl/acl_family.h

@@ -30,11 +30,15 @@ class AclFamily final {
   void Acl(CmdArgList args, ConnectionContext* cntx);
   void List(CmdArgList args, ConnectionContext* cntx);
   void SetUser(CmdArgList args, ConnectionContext* cntx);
+  void DelUser(CmdArgList args, ConnectionContext* cntx);
 
   // Helper function that updates all open connections and their
   // respective ACL fields on all the available proactor threads
   void StreamUpdatesToAllProactorConnections(std::string_view user, uint32_t update_cat);
 
+  // Helper function that closes all open connection from the deleted user
+  void EvictOpenConnectionsOnAllProactors(std::string_view user);
+
   facade::Listener* main_listener_{nullptr};
 };
 

src/server/acl/user_registry.cc

@@ -21,10 +21,9 @@ void UserRegistry::MaybeAddAndUpdate(std::string_view username, User::UpdateRequ
   user.Update(std::move(req));
 }
 
-void UserRegistry::RemoveUser(std::string_view username) {
+bool UserRegistry::RemoveUser(std::string_view username) {
   std::unique_lock<util::SharedMutex> lock(mu_);
-  registry_.erase(username);
-  // TODO evict authed connections from user
+  return registry_.erase(username);
 }
 
 UserRegistry::UserCredentials UserRegistry::GetCredentials(std::string_view username) const {

src/server/acl/user_registry.h

@@ -35,8 +35,7 @@ class UserRegistry {
   // Acquires a write lock on mu_
   // Removes user from the store
   // kills already existing connections from the removed user
-  // TODO change return time to communicate back results to acl commands
-  void RemoveUser(std::string_view username);
+  bool RemoveUser(std::string_view username);
 
   struct UserCredentials {
     uint32_t acl_categories{0};

tests/dragonfly/acl_family_test.py

@@ -3,7 +3,6 @@
 from redis import asyncio as aioredis
 from . import DflyInstanceFactory
 from .utility import disconnect_clients
-import time
 import asyncio
 
 
@@ -193,6 +192,33 @@ async def test_acl_categories_multi_exec_squash(df_local_factory):
     await client.close()
 
 
+@pytest.mark.asyncio
+async def test_acl_deluser(df_server):
+    client = aioredis.Redis(port=df_server.port)
+
+    try:
+        res = await client.execute_command("ACL DELUSER adi")
+    except redis.exceptions.ResponseError as e:
+        assert e.args[0] == "User adi does not exist"
+
+    res = await client.execute_command("ACL SETUSER adi ON >pass +@transaction +@string")
+    assert res == b"OK"
+
+    res = await client.execute_command("AUTH adi pass")
+    assert res == b"OK"
+
+    await client.execute_command("MULTI")
+    await client.execute_command("SET key 44")
+
+    admin_client = aioredis.Redis(port=df_server.port)
+    await admin_client.execute_command("ACL DELUSER adi")
+
+    with pytest.raises(redis.exceptions.ConnectionError):
+        await client.execute_command("EXEC")
+
+    await admin_client.close()
+
+
 script = """
 for i = 1, 10000 do
   redis.call('SET', 'key', i)
@@ -203,6 +229,26 @@ async def test_acl_categories_multi_exec_squash(df_local_factory):
 """
 
 
+@pytest.mark.asyncio
+async def test_acl_del_user_while_running_lua_script(df_server):
+    client = aioredis.Redis(port=df_server.port)
+    await client.execute_command("ACL SETUSER kostas ON >kk +@string +@scripting")
+    await client.execute_command("AUTH kostas kk")
+    admin_client = aioredis.Redis(port=df_server.port)
+
+    with pytest.raises(redis.exceptions.ConnectionError):
+        await asyncio.gather(
+            client.eval(script, 4, "key", "key1", "key2", "key3"),
+            admin_client.execute_command("ACL DELUSER kostas"),
+        )
+
+    for i in range(1, 4):
+        res = await admin_client.get(f"key{i}")
+        assert res == b"10000"
+
+    await admin_client.close()
+
+
 @pytest.mark.asyncio
 async def test_acl_with_long_running_script(df_server):
     client = aioredis.Redis(port=df_server.port)