Free current client asynchronously after user permissions changes (#13274)

The crash happens when the user that triggers the permission changes
should be affected (and should be disconnected eventually).

To handle such a scenario, we should use the
`CLIENT_CLOSE_AFTER_COMMAND` flag.

This commit encapsulates all the places that should be handled in the
same way in `deauthenticateAndCloseClient`

Also:
* bugfix: during the ACL LOAD we ignore clients that are marked as
`CLIENT MASTER`

src/acl.c

@@ -485,15 +485,7 @@ void ACLFreeUserAndKillClients(user *u) {
              * this may result in some security hole: it's much
              * more defensive to set the default user and put
              * it in non authenticated mode. */
-            c->user = DefaultUser;
+            deauthenticateAndCloseClient(c);
-            c->authenticated = 0;
-            /* We will write replies to this client later, so we can't
-             * close it directly even if async. */
-            if (c == server.current_client) {
-                c->flags |= CLIENT_CLOSE_AFTER_COMMAND;
-            } else {
-                freeClientAsync(c);
-            }
         }
     }
     ACLFreeUser(u);
@@ -2010,7 +2002,7 @@ void ACLKillPubsubClientsIfNeeded(user *new, user *original) {
         if (c->user != original)
             continue;
         if (ACLShouldKillPubsubClient(c, channels))
-            freeClient(c);
+            deauthenticateAndCloseClient(c);
     }
 
     listRelease(channels);
@@ -2433,6 +2425,9 @@ sds ACLLoadFromFile(const char *filename) {
         listRewind(server.clients,&li);
         while ((ln = listNext(&li)) != NULL) {
             client *c = listNodeValue(ln);
+            /* a MASTER client can do everything (and user = NULL) so we can skip it */
+            if (c->flags & CLIENT_MASTER)
+                continue;
             user *original = c->user;
             list *channels = NULL;
             user *new = ACLGetUserByName(c->user->name, sdslen(c->user->name));
@@ -2444,7 +2439,7 @@ sds ACLLoadFromFile(const char *filename) {
             }
             /* When the new channel list is NULL, it means the new user's channel list is a superset of the old user's list. */
             if (!new || (channels && ACLShouldKillPubsubClient(c, channels))) {
-                freeClient(c);
+                deauthenticateAndCloseClient(c);
                 continue;
             }
             c->user = new;

src/module.c

@@ -9548,15 +9548,7 @@ void revokeClientAuthentication(client *c) {
      * is eventually freed we don't rely on the module to still exist. */
     moduleNotifyUserChanged(c);
 
-    c->user = DefaultUser;
+    deauthenticateAndCloseClient(c);
-    c->authenticated = 0;
-    /* We will write replies to this client later, so we can't close it
-     * directly even if async. */
-    if (c == server.current_client) {
-        c->flags |= CLIENT_CLOSE_AFTER_COMMAND;
-    } else {
-        freeClientAsync(c);
-    }
 }
 
 /* Cleanup all clients that have been authenticated with this module. This

src/networking.c

@@ -1549,6 +1549,18 @@ void clearClientConnectionState(client *c) {
                   CLIENT_REPLY_SKIP_NEXT|CLIENT_NO_TOUCH|CLIENT_NO_EVICT);
 }
 
+void deauthenticateAndCloseClient(client *c) {
+    c->user = DefaultUser;
+    c->authenticated = 0;
+    /* We will write replies to this client later, so we can't
+     * close it directly even if async. */
+    if (c == server.current_client) {
+        c->flags |= CLIENT_CLOSE_AFTER_COMMAND;
+    } else {
+        freeClientAsync(c);
+    }
+}
+
 void freeClient(client *c) {
     listNode *ln;
 

src/server.h

@@ -2571,6 +2571,7 @@ void redisSetCpuAffinity(const char *cpulist);
 client *createClient(connection *conn);
 void freeClient(client *c);
 void freeClientAsync(client *c);
+void deauthenticateAndCloseClient(client *c);
 void logInvalidUseAndFreeClientAsync(client *c, const char *fmt, ...);
 int beforeNextClient(client *c);
 void clearClientConnectionState(client *c);

tests/unit/acl.tcl

@@ -247,6 +247,26 @@ start_server {tags {"acl external:skip"}} {
         set e
     } {*NOPERM*channel*}
 
+    test {Subscribers are killed when revoked of channel permission} {
+        # This test covers the case that the SETUSER is requested over the subscriber
+        set rd [redis_deferring_client]
+        r ACL setuser psuser resetchannels &foo:1
+        # we must use RESP 3 since AUTH command is not supported over a subscribed client with RESP2
+        $rd HELLO 3 AUTH psuser pspass
+        $rd read
+        $rd CLIENT SETNAME deathrow
+        $rd read
+        $rd SUBSCRIBE foo:1
+        assert_match {subscribe foo:1 1} [$rd read]
+        $rd ACL setuser psuser resetchannels
+        assert_match {OK} [$rd read]
+        # 'psuser' no longer has access to "foo:1" channel, so they should get disconnected
+        catch {$rd read} e
+        assert_match {*I/O error*} $e
+        assert_no_match {*deathrow*} [r CLIENT LIST]
+        $rd close
+    } {0}
+
     test {Subscribers are killed when revoked of channel permission} {
         set rd [redis_deferring_client]
         r ACL setuser psuser resetchannels &foo:1
@@ -1051,6 +1071,29 @@ start_server [list overrides [list "dir" $server_path "acl-pubsub-default" "allc
         $rd2 close
     }
 
+    test {ACL LOAD disconnects affected subscriber} {
+        # This test covers the case that the LOAD is requested over the subscriber
+        reconnect
+        r ACL SETUSER doug on nopass resetchannels &test* +@all ~*
+
+        set rd1 [redis_deferring_client]
+
+        # we must use RESP 3 since AUTH command is not supported over a subscribed client with RESP2
+        $rd1 HELLO 3 AUTH doug doug
+        $rd1 read
+        $rd1 SUBSCRIBE test1
+        $rd1 read
+
+        $rd1 ACL LOAD
+        assert_match {OK} [$rd1 read]
+
+        # 'doug' no longer has access to "test1" channel, so they should get disconnected
+        catch {$rd1 read} e
+        assert_match {*I/O error*} $e
+
+        $rd1 close
+    }
+
     test {ACL LOAD disconnects clients of deleted users} {
         reconnect
         r ACL SETUSER mortimer on >mortimer ~* &* +@all
@@ -1113,6 +1156,28 @@ start_server [list overrides [list "dir" $server_path "acl-pubsub-default" "allc
         r ACL deluser harry
         set e
     } {*NOPERM*channel*}
+
+    set server_path [tmpdir "server.acl"]
+    exec cp -f tests/assets/user.acl $server_path
+    start_server [list overrides [list "dir" $server_path "acl-pubsub-default" "allchannels" "aclfile" "user.acl"] tags [list "repl" "external:skip"]] {
+        set master [srv -1 client]
+        set master_host [srv -1 host]
+        set master_port [srv -1 port]
+        set slave [srv 0 client]
+
+        test {First server should have role slave after SLAVEOF} {
+            $slave slaveof $master_host $master_port
+            wait_for_condition 50 100 {
+                [s 0 master_link_status] eq {up}
+            } else {
+                fail "Replication not started."
+            }
+        }
+
+        test {ACL load on replica when connected to replica} {
+            assert_match {OK} [$slave ACL LOAD]
+        }
+    }
 }
 
 set server_path [tmpdir "resetchannels.acl"]