Added a shared secret over Redis cluster. (#13763)

The PR introduces a new shared secret that is shared over all the nodes
on the Redis cluster. The main idea is to leverage the cluster bus to
share a secret between all the nodes such that later the nodes will be
able to authenticate using this secret and send internal commands to
each other (see #13740 for more information about internal commands).

The way the shared secret is chosen is the following:
1. Each node, when start, randomly generate its own internal secret.
2. Each node share its internal secret over the cluster ping messages.
3. If a node gets a ping message with secret smaller then his current
secret, it embrace it.
4. Eventually all nodes should embrace the minimal secret

The converges of the secret is as good as the topology converges.

To extend the ping messages to contain the secret, we leverage the
extension mechanism. Nodes that runs an older Redis version will just
ignore those extensions.

Specific tests were added to verify that eventually all nodes see the
secrets. In addition, a verification was added to the test infra to
verify the secret on `cluster_config_consistent` and to
`assert_cluster_state`.

src/cluster.h

@@ -131,6 +131,7 @@ char *clusterNodeHostname(clusterNode *node);
 const char *clusterNodePreferredEndpoint(clusterNode *n);
 long long clusterNodeReplOffset(clusterNode *node);
 clusterNode *clusterLookupNode(const char *name, int length);
+const char *clusterGetSecret(size_t *len);
 
 /* functions with shared implementations */
 clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, int argc, int *hashslot, uint64_t cmd_flags, int *error_code);

src/cluster_legacy.c

@@ -1030,6 +1030,8 @@ void clusterInit(void) {
     clusterUpdateMyselfIp();
     clusterUpdateMyselfHostname();
     clusterUpdateMyselfHumanNodename();
+
+    getRandomHexChars(server.cluster->internal_secret, CLUSTER_INTERNALSECRETLEN);
 }
 
 void clusterInitLast(void) {
@@ -1579,6 +1581,14 @@ clusterNode *clusterLookupNode(const char *name, int length) {
     return dictGetVal(de);
 }
 
+const char *clusterGetSecret(size_t *len) {
+    if (!server.cluster) {
+        return NULL;
+    }
+    *len = CLUSTER_INTERNALSECRETLEN;
+    return server.cluster->internal_secret;
+}
+
 /* Get all the nodes in my shard.
  * Note that the list returned is not computed on the fly
  * via slaveof; rather, it is maintained permanently to
@@ -2503,6 +2513,10 @@ uint32_t getShardIdPingExtSize(void) {
     return getAlignedPingExtSize(sizeof(clusterMsgPingExtShardId));
 }
 
+uint32_t getInternalSecretPingExtSize(void) {
+    return getAlignedPingExtSize(sizeof(clusterMsgPingExtInternalSecret));
+}
+
 uint32_t getForgottenNodeExtSize(void) {
     return getAlignedPingExtSize(sizeof(clusterMsgPingExtForgottenNode));
 }
@@ -2594,6 +2608,17 @@ uint32_t writePingExt(clusterMsg *hdr, int gossipcount)  {
     totlen += getShardIdPingExtSize();
     extensions++;
 
+    /* Populate insternal secret */
+    if (cursor != NULL) {
+        clusterMsgPingExtInternalSecret *ext = preparePingExt(cursor, CLUSTERMSG_EXT_TYPE_INTERNALSECRET, getInternalSecretPingExtSize());
+        memcpy(ext->internal_secret, server.cluster->internal_secret, CLUSTER_INTERNALSECRETLEN);
+
+        /* Move the write cursor */
+        cursor = nextPingExt(cursor);
+    }
+    totlen += getInternalSecretPingExtSize();
+    extensions++;
+
     if (hdr != NULL) {
         hdr->extensions = htons(extensions);
     }
@@ -2634,9 +2659,14 @@ void clusterProcessPingExtensions(clusterMsg *hdr, clusterLink *link) {
         } else if (type == CLUSTERMSG_EXT_TYPE_SHARDID) {
             clusterMsgPingExtShardId *shardid_ext = (clusterMsgPingExtShardId *) &(ext->ext[0].shard_id);
             ext_shardid = shardid_ext->shard_id;
+        } else if (type == CLUSTERMSG_EXT_TYPE_INTERNALSECRET) {
+            clusterMsgPingExtInternalSecret *internal_secret_ext = (clusterMsgPingExtInternalSecret *) &(ext->ext[0].internal_secret);
+            if (memcmp(server.cluster->internal_secret, internal_secret_ext->internal_secret, CLUSTER_INTERNALSECRETLEN) > 0 ) {
+                memcpy(server.cluster->internal_secret, internal_secret_ext->internal_secret, CLUSTER_INTERNALSECRETLEN);
+            }
         } else {
             /* Unknown type, we will ignore it but log what happened. */
-            serverLog(LL_WARNING, "Received unknown extension type %d", type);
+            serverLog(LL_VERBOSE, "Received unknown extension type %d", type);
         }
 
         /* We know this will be valid since we validated it ahead of time */

src/cluster_legacy.h

@@ -148,10 +148,12 @@ typedef enum {
     CLUSTERMSG_EXT_TYPE_HUMAN_NODENAME,
     CLUSTERMSG_EXT_TYPE_FORGOTTEN_NODE,
     CLUSTERMSG_EXT_TYPE_SHARDID,
+    CLUSTERMSG_EXT_TYPE_INTERNALSECRET,
 } clusterMsgPingtypes;
 
 /* Helper function for making sure extensions are eight byte aligned. */
 #define EIGHT_BYTE_ALIGN(size) ((((size) + 7) / 8) * 8)
+#define CLUSTER_INTERNALSECRETLEN 40      /* sha1 hex length */
 
 typedef struct {
     char hostname[1]; /* The announced hostname, ends with \0. */
@@ -172,6 +174,10 @@ typedef struct {
     char shard_id[CLUSTER_NAMELEN]; /* The shard_id, 40 bytes fixed. */
 } clusterMsgPingExtShardId;
 
+typedef struct {
+    char internal_secret[CLUSTER_INTERNALSECRETLEN]; /* Current shard internal secret */
+} clusterMsgPingExtInternalSecret;
+
 typedef struct {
     uint32_t length; /* Total length of this extension message (including this header) */
     uint16_t type; /* Type of this extension message (see clusterMsgPingExtTypes) */
@@ -181,6 +187,7 @@ typedef struct {
         clusterMsgPingExtHumanNodename human_nodename;
         clusterMsgPingExtForgottenNode forgotten_node;
         clusterMsgPingExtShardId shard_id;
+        clusterMsgPingExtInternalSecret internal_secret;
     } ext[]; /* Actual extension information, formatted so that the data is 8
               * byte aligned, regardless of its content. */
 } clusterMsgPingExt;
@@ -333,6 +340,7 @@ struct clusterState {
     clusterNode *migrating_slots_to[CLUSTER_SLOTS];
     clusterNode *importing_slots_from[CLUSTER_SLOTS];
     clusterNode *slots[CLUSTER_SLOTS];
+    char internal_secret[CLUSTER_INTERNALSECRETLEN];
     /* The following fields are used to take the slave state on elections. */
     mstime_t failover_auth_time; /* Time of previous or next election. */
     int failover_auth_count;    /* Number of votes received so far. */

src/debug.c

@@ -397,6 +397,8 @@ void debugCommand(client *c) {
 "    Hard crash and restart after a <milliseconds> delay (default 0).",
 "DIGEST",
 "    Output a hex signature representing the current DB content.",
+"INTERNAL_SECRET",
+"    Return the cluster internal secret (hashed with crc16) or error if not in cluster mode.",
 "DIGEST-VALUE <key> [<key> ...]",
 "    Output a hex signature of the values of all the specified keys.",
 "ERROR <string>",
@@ -759,6 +761,15 @@ NULL
         for (int i = 0; i < 20; i++) d = sdscatprintf(d, "%02x",digest[i]);
         addReplyStatus(c,d);
         sdsfree(d);
+    } else if (!strcasecmp(c->argv[1]->ptr,"internal_secret") && c->argc == 2) {
+        size_t len;
+        const char *internal_secret = clusterGetSecret(&len);
+        if (!internal_secret) {
+            addReplyError(c, "Internal secret is missing");
+        } else {
+            uint16_t hash = crc16(internal_secret, len);
+            addReplyLongLong(c, hash);
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"digest-value") && c->argc >= 2) {
         /* DEBUG DIGEST-VALUE key key key ... key. */
         addReplyArrayLen(c,c->argc-2);

tests/cluster/cluster.tcl

@@ -96,6 +96,31 @@ proc assert_cluster_state {state} {
             fail "Cluster node $id cluster_state:[CI $id cluster_state]"
         }
     }
+
+    wait_for_secrets_match 50 100
+}
+
+proc num_unique_secrets {} {
+    set secrets [list]
+    foreach_redis_id id {
+        if {[instance_is_killed redis $id]} continue
+        lappend secrets [R $id debug internal_secret]
+    }
+    set num_secrets [llength [lsort -unique $secrets]]
+    return $num_secrets
+}
+
+# Check that cluster nodes agree about "state", or raise an error.
+proc assert_secrets_match {} {
+    assert_equal {1} [num_unique_secrets]
+}
+
+proc wait_for_secrets_match {maxtries delay} {
+    wait_for_condition $maxtries $delay {
+        [num_unique_secrets] eq 1
+    } else {
+        fail "Failed waiting for secrets to sync"
+    }
 }
 
 # Search the first node starting from ID $first that is not
@@ -188,9 +213,11 @@ proc cluster_config_consistent {} {
     for {set j 0} {$j < $::cluster_master_nodes + $::cluster_replica_nodes} {incr j} {
         if {$j == 0} {
             set base_cfg [R $j cluster slots]
+            set base_secret [R $j debug internal_secret]
         } else {
             set cfg [R $j cluster slots]
-            if {$cfg != $base_cfg} {
+            set secret [R $j debug internal_secret]
+            if {$cfg != $base_cfg || $secret != $base_secret} {
                 return 0
             }
         }

tests/support/cluster_util.tcl

@@ -5,8 +5,9 @@ proc cluster_config_consistent {} {
     for {set j 0} {$j < [llength $::servers]} {incr j} {
         if {$j == 0} {
             set base_cfg [R $j cluster slots]
+            set base_secret [R $j debug internal_secret]
         } else {
-            if {[R $j cluster slots] != $base_cfg} {
+            if {[R $j cluster slots] != $base_cfg || [R $j debug internal_secret] != $base_secret} {
                 return 0
             }
         }

tests/unit/cluster/internal-secret.tcl

@@ -0,0 +1,71 @@
+proc num_unique_secrets {num_nodes} {
+    set secrets [list]
+    for {set i 0} {$i < $num_nodes} {incr i} {
+        lappend secrets [R $i debug internal_secret]
+    }
+    set num_secrets [llength [lsort -unique $secrets]]
+    return $num_secrets
+}
+
+proc wait_for_secret_sync {maxtries delay num_nodes} {
+    wait_for_condition $maxtries $delay {
+        [num_unique_secrets $num_nodes] eq 1
+    } else {
+        fail "Failed waiting for secrets to sync"
+    }
+}
+
+start_cluster 10 10 {tags {external:skip cluster}} {
+    test "Test internal secret sync" {
+        wait_for_secret_sync 50 100 20
+    }
+
+    
+    set first_shard_host [srv 0 host]
+    set first_shard_port [srv 0 port]
+    
+    if {$::verbose} {
+        puts {cluster internal secret:}
+        puts [R 1 debug internal_secret]
+    }
+
+    test "Join a node to the cluster and make sure it gets the same secret" {
+        start_server {tags {"external:skip"} overrides {cluster-enabled {yes}}} {
+            r cluster meet $first_shard_host $first_shard_port
+            wait_for_condition 50 100 {
+                [r debug internal_secret] eq [R 1 debug internal_secret]
+            } else {
+                puts [r debug internal_secret]
+                puts [R 1 debug internal_secret]
+                fail "Secrets not match"
+            }
+        }
+    }
+
+    test "Join another cluster, make sure clusters sync on the internal secret" {
+        start_server {tags {"external:skip"} overrides {cluster-enabled {yes}}} {
+            set new_shard_host [srv 0 host]
+            set new_shard_port [srv 0 port]
+            start_server {tags {"external:skip"} overrides {cluster-enabled {yes}}} {
+                r cluster meet $new_shard_host $new_shard_port
+                wait_for_condition 50 100 {
+                    [r debug internal_secret] eq [r -1 debug internal_secret]
+                } else {
+                    puts [r debug internal_secret]
+                    puts [r -1 debug internal_secret]
+                    fail "Secrets not match"
+                }
+                if {$::verbose} {
+                    puts {new cluster internal secret:}
+                    puts [r -1 debug internal_secret]
+                }
+                r cluster meet $first_shard_host $first_shard_port
+                wait_for_secret_sync 50 100 22
+                if {$::verbose} {
+                    puts {internal secret after join to bigger cluster:}
+                    puts [r -1 debug internal_secret]
+                }
+            }
+        }
+    }
+}