Added rd_kafka_sasl_set_credentials() (#4033)

josvisser66 · edenhill · web-flow · commit 3b50e1eb5feb · 2022-11-09T09:28:58.000+01:00
Co-authored-by: Jos Visser &lt;jvisser@confluent.io&gt;
Co-authored-by: edenhill &lt;magnus@edenhill.se&gt;

Co-authored-by: edenhill &lt;magnus@edenhill.se&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -42,6 +42,7 @@ configuration property.
  * Bundled zlib upgraded to version 1.2.13.
  * Added `on_broker_state_change()` interceptor
  * The C++ API no longer returns strings by const value, which enables better move optimization in callers.
+ * Added `rd_kafka_sasl_set_credentials()` API to update SASL credentials.
  * Setting `allow.auto.create.topics` will no longer give a warning if used by a producer, since that is an expected use case.
   Improvement in documentation for this property.
 
@@ -65,7 +66,6 @@ configuration property.
 
  * Back-off and retry JoinGroup request if coordinator load is in progress.
 
-
 # librdkafka v1.9.2
 
 librdkafka v1.9.2 is a maintenance release:
diff --git a/src-cpp/rdkafkacpp.h b/src-cpp/rdkafkacpp.h
@@ -1899,6 +1899,23 @@ class RD_EXPORT Handle {
    *         that explicitly mention using this function for freeing.
    */
   virtual void mem_free(void *ptr) = 0;
+
+  /**
+   * @brief Sets SASL credentials used for SASL PLAIN and SCRAM mechanisms by
+   *        this Kafka client.
+   *
+   * This function sets or resets the SASL username and password credentials
+   * used by this Kafka client. The new credentials will be used the next time
+   * this client needs to authenticate to a broker.
+   * will not disconnect existing connections that might have been made using
+   * the old credentials.
+   *
+   * @remark This function only applies to the SASL PLAIN and SCRAM mechanisms.
+   *
+   * @returns NULL on success or an error object on error.
+   */
+  virtual Error *sasl_set_credentials(const std::string &username,
+                                      const std::string &password) = 0;
 };
 
 
diff --git a/src-cpp/rdkafkacpp_int.h b/src-cpp/rdkafkacpp_int.h
@@ -1144,6 +1144,17 @@ class HandleImpl : virtual public Handle {
     return NULL;
   }
 
+  Error *sasl_set_credentials(const std::string &username,
+                              const std::string &password) {
+    rd_kafka_error_t *c_error =
+        rd_kafka_sasl_set_credentials(rk_, username.c_str(), password.c_str());
+
+    if (c_error)
+      return new ErrorImpl(c_error);
+
+    return NULL;
+  };
+
   void *mem_malloc(size_t size) {
     return rd_kafka_mem_malloc(rk_, size);
   }
diff --git a/src/rdkafka.c b/src/rdkafka.c
@@ -995,6 +995,7 @@ void rd_kafka_destroy_final(rd_kafka_t *rk) {
         rd_kafka_anyconf_destroy(_RK_GLOBAL, &rk->rk_conf);
         rd_list_destroy(&rk->rk_broker_by_id);
 
+        mtx_destroy(&rk->rk_conf.sasl.lock);
         rwlock_destroy(&rk->rk_lock);
 
         rd_free(rk);
@@ -2205,6 +2206,7 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
         rd_kafka_interceptors_on_new(rk, &rk->rk_conf);
 
         rwlock_init(&rk->rk_lock);
+        mtx_init(&rk->rk_conf.sasl.lock, mtx_plain);
         mtx_init(&rk->rk_internal_rkb_lock, mtx_plain);
 
         cnd_init(&rk->rk_broker_state_change_cnd);
diff --git a/src/rdkafka.h b/src/rdkafka.h
@@ -3331,6 +3331,25 @@ RD_EXPORT
 rd_kafka_error_t *rd_kafka_sasl_background_callbacks_enable(rd_kafka_t *rk);
 
 
+/**
+ * @brief Sets SASL credentials used for SASL PLAIN and SCRAM mechanisms by
+ *        this Kafka client.
+ *
+ * This function sets or resets the SASL username and password credentials
+ * used by this Kafka client. The new credentials will be used the next time
+ * this client needs to authenticate to a broker. This function
+ * will not disconnect existing connections that might have been made using
+ * the old credentials.
+ *
+ * @remark This function only applies to the SASL PLAIN and SCRAM mechanisms.
+ *
+ * @returns NULL on success or an error object on error.
+ */
+RD_EXPORT
+rd_kafka_error_t *rd_kafka_sasl_set_credentials(rd_kafka_t *rk,
+                                                const char *username,
+                                                const char *password);
+
 /**
  * @returns a reference to the librdkafka consumer queue.
  * This is the queue served by rd_kafka_consumer_poll().
diff --git a/src/rdkafka_conf.h b/src/rdkafka_conf.h
@@ -159,7 +159,7 @@ typedef enum {
 
 /* Increase in steps of 64 as needed.
  * This must be larger than sizeof(rd_kafka_[topic_]conf_t) */
-#define RD_KAFKA_CONF_PROPS_IDX_MAX (64 * 31)
+#define RD_KAFKA_CONF_PROPS_IDX_MAX (64 * 33)
 
 /**
  * @struct rd_kafka_anyconf_t
@@ -276,6 +276,9 @@ struct rd_kafka_conf_s {
                 char *kinit_cmd;
                 char *keytab;
                 int relogin_min_time;
+                /** Protects .username and .password access after client
+                 *  instance has been created (see sasl_set_credentials()). */
+                mtx_t lock;
                 char *username;
                 char *password;
 #if WITH_SASL_SCRAM
diff --git a/src/rdkafka_sasl.c b/src/rdkafka_sasl.c
@@ -488,3 +488,35 @@ int rd_kafka_sasl_global_init(void) {
         return 0;
 #endif
 }
+
+/**
+ * Sets or resets the SASL (PLAIN or SCRAM) credentials used by this
+ * client when making new connections to brokers.
+ *
+ * @returns NULL on success or an error object on error.
+ */
+rd_kafka_error_t *rd_kafka_sasl_set_credentials(rd_kafka_t *rk,
+                                                const char *username,
+                                                const char *password) {
+
+        if (!username || !password)
+                return rd_kafka_error_new(RD_KAFKA_RESP_ERR__INVALID_ARG,
+                                          "Username and password are required");
+
+        mtx_lock(&rk->rk_conf.sasl.lock);
+
+        if (rk->rk_conf.sasl.username)
+                rd_free(rk->rk_conf.sasl.username);
+        rk->rk_conf.sasl.username = rd_strdup(username);
+
+        if (rk->rk_conf.sasl.password)
+                rd_free(rk->rk_conf.sasl.password);
+        rk->rk_conf.sasl.password = rd_strdup(password);
+
+        mtx_unlock(&rk->rk_conf.sasl.lock);
+
+        rd_kafka_all_brokers_wakeup(rk, RD_KAFKA_BROKER_STATE_INIT,
+                                    "SASL credentials updated");
+
+        return NULL;
+}
diff --git a/src/rdkafka_sasl_cyrus.c b/src/rdkafka_sasl_cyrus.c
@@ -91,8 +91,10 @@ static int rd_kafka_sasl_cyrus_recv(struct rd_kafka_transport_s *rktrans,
                 const char *out;
                 unsigned int outlen;
 
+                mtx_lock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
                 r = sasl_client_step(state->conn, size > 0 ? buf : NULL, size,
                                      &interact, &out, &outlen);
+                mtx_unlock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
 
                 if (r >= 0) {
                         /* Note: outlen may be 0 here for an empty response */
@@ -148,9 +150,11 @@ static int rd_kafka_sasl_cyrus_recv(struct rd_kafka_transport_s *rktrans,
             RD_KAFKA_DBG_SECURITY) {
                 const char *user, *mech, *authsrc;
 
+                mtx_lock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
                 if (sasl_getprop(state->conn, SASL_USERNAME,
                                  (const void **)&user) != SASL_OK)
                         user = "(unknown)";
+                mtx_unlock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
 
                 if (sasl_getprop(state->conn, SASL_MECHNAME,
                                  (const void **)&mech) != SASL_OK)
@@ -356,6 +360,12 @@ static int rd_kafka_sasl_cyrus_cb_getsimple(void *context,
         switch (id) {
         case SASL_CB_USER:
         case SASL_CB_AUTHNAME:
+                /* Since cyrus expects the returned pointer to be stable
+                 * and not have its content changed, but the username
+                 * and password may be updated at anytime by the application
+                 * calling sasl_set_credentials(), we need to lock
+                 * rk_conf.sasl.lock before each call into cyrus-sasl.
+                 * So when we get here the lock is already held. */
                 *result = rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.username;
                 break;
 
@@ -381,6 +391,7 @@ static int rd_kafka_sasl_cyrus_cb_getsecret(sasl_conn_t *conn,
         rd_kafka_transport_t *rktrans = context;
         const char *password;
 
+        /* rk_conf.sasl.lock is already locked */
         password = rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.password;
 
         if (!password) {
@@ -472,8 +483,11 @@ static void rd_kafka_sasl_cyrus_close(struct rd_kafka_transport_s *rktrans) {
         if (!state)
                 return;
 
-        if (state->conn)
+        if (state->conn) {
+                mtx_lock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
                 sasl_dispose(&state->conn);
+                mtx_unlock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
+        }
         rd_free(state);
 }
 
@@ -528,9 +542,11 @@ static int rd_kafka_sasl_cyrus_client_new(rd_kafka_transport_t *rktrans,
 
         memcpy(state->callbacks, callbacks, sizeof(callbacks));
 
+        mtx_lock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
         r = sasl_client_new(rk->rk_conf.sasl.service_name, hostname, NULL,
                             NULL, /* no local & remote IP checks */
                             state->callbacks, 0, &state->conn);
+        mtx_unlock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
         if (r != SASL_OK) {
                 rd_snprintf(errstr, errstr_size, "%s",
                             sasl_errstring(r, NULL, NULL));
@@ -550,8 +566,10 @@ static int rd_kafka_sasl_cyrus_client_new(rd_kafka_transport_t *rktrans,
                 unsigned int outlen;
                 const char *mech = NULL;
 
+                mtx_lock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
                 r = sasl_client_start(state->conn, rk->rk_conf.sasl.mechanisms,
                                       NULL, &out, &outlen, &mech);
+                mtx_unlock(&rktrans->rktrans_rkb->rkb_rk->rk_conf.sasl.lock);
 
                 if (r >= 0)
                         if (rd_kafka_sasl_send(rktrans, out, outlen, errstr,
diff --git a/src/rdkafka_sasl_plain.c b/src/rdkafka_sasl_plain.c
@@ -74,13 +74,16 @@ int rd_kafka_sasl_plain_client_new(rd_kafka_transport_t *rktrans,
         char *buf;
         int of     = 0;
         int zidlen = 0;
-        int cidlen = rk->rk_conf.sasl.username
-                         ? (int)strlen(rk->rk_conf.sasl.username)
-                         : 0;
-        int pwlen = rk->rk_conf.sasl.password
-                        ? (int)strlen(rk->rk_conf.sasl.password)
-                        : 0;
+        int cidlen, pwlen;
 
+        mtx_lock(&rk->rk_conf.sasl.lock);
+
+        cidlen = rk->rk_conf.sasl.username
+                     ? (int)strlen(rk->rk_conf.sasl.username)
+                     : 0;
+        pwlen = rk->rk_conf.sasl.password
+                    ? (int)strlen(rk->rk_conf.sasl.password)
+                    : 0;
 
         buf = rd_alloca(zidlen + 1 + cidlen + 1 + pwlen + 1);
 
@@ -95,6 +98,7 @@ int rd_kafka_sasl_plain_client_new(rd_kafka_transport_t *rktrans,
         /* passwd */
         memcpy(&buf[of], rk->rk_conf.sasl.password, pwlen);
         of += pwlen;
+        mtx_unlock(&rk->rk_conf.sasl.lock);
 
         rd_rkb_dbg(rkb, SECURITY, "SASLPLAIN",
                    "Sending SASL PLAIN (builtin) authentication token");
@@ -115,7 +119,13 @@ int rd_kafka_sasl_plain_client_new(rd_kafka_transport_t *rktrans,
 static int rd_kafka_sasl_plain_conf_validate(rd_kafka_t *rk,
                                              char *errstr,
                                              size_t errstr_size) {
-        if (!rk->rk_conf.sasl.username || !rk->rk_conf.sasl.password) {
+        rd_bool_t both_set;
+
+        mtx_lock(&rk->rk_conf.sasl.lock);
+        both_set = rk->rk_conf.sasl.username && rk->rk_conf.sasl.password;
+        mtx_unlock(&rk->rk_conf.sasl.lock);
+
+        if (!both_set) {
                 rd_snprintf(errstr, errstr_size,
                             "sasl.username and sasl.password must be set");
                 return -1;
diff --git a/src/rdkafka_sasl_scram.c b/src/rdkafka_sasl_scram.c
@@ -397,9 +397,8 @@ static int rd_kafka_sasl_scram_build_client_final_message(
     int itcnt,
     rd_chariov_t *out) {
         struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state;
-        const rd_kafka_conf_t *conf  = &rktrans->rktrans_rkb->rkb_rk->rk_conf;
-        rd_chariov_t SaslPassword    = {.ptr  = conf->sasl.password,
-                                     .size = strlen(conf->sasl.password)};
+        rd_kafka_conf_t *conf        = &rktrans->rktrans_rkb->rkb_rk->rk_conf;
+        rd_chariov_t SaslPassword    = RD_ZERO_INIT;
         rd_chariov_t SaltedPassword  = {.ptr = rd_alloca(EVP_MAX_MD_SIZE)};
         rd_chariov_t ClientKey       = {.ptr = rd_alloca(EVP_MAX_MD_SIZE)};
         rd_chariov_t ServerKey       = {.ptr = rd_alloca(EVP_MAX_MD_SIZE)};
@@ -416,6 +415,11 @@ static int rd_kafka_sasl_scram_build_client_final_message(
         char *ClientProofB64;
         int i;
 
+        mtx_lock(&conf->sasl.lock);
+        rd_strdupa(&SaslPassword.ptr, conf->sasl.password);
+        mtx_unlock(&conf->sasl.lock);
+        SaslPassword.size = strlen(SaslPassword.ptr);
+
         /* Constructing the ClientProof attribute (p):
          *
          * p = Base64-encoded ClientProof
@@ -664,7 +668,7 @@ rd_kafka_sasl_scram_handle_server_final_message(rd_kafka_transport_t *rktrans,
         } else if ((attr_v = rd_kafka_sasl_scram_get_attr(
                         in, 'v', "verifier in server-final-message", errstr,
                         errstr_size))) {
-                const rd_kafka_conf_t *conf;
+                rd_kafka_conf_t *conf;
 
                 /* Authentication succesful on server,
                  * but we need to verify the ServerSignature too. */
@@ -686,9 +690,11 @@ rd_kafka_sasl_scram_handle_server_final_message(rd_kafka_transport_t *rktrans,
 
                 conf = &rktrans->rktrans_rkb->rkb_rk->rk_conf;
 
+                mtx_lock(&conf->sasl.lock);
                 rd_rkb_dbg(rktrans->rktrans_rkb, SECURITY | RD_KAFKA_DBG_BROKER,
                            "SCRAMAUTH", "Authenticated as %s using %s",
                            conf->sasl.username, conf->sasl.mechanisms);
+                mtx_unlock(&conf->sasl.lock);
 
                 rd_kafka_sasl_auth_done(rktrans);
                 return 0;
@@ -711,11 +717,13 @@ rd_kafka_sasl_scram_build_client_first_message(rd_kafka_transport_t *rktrans,
                                                rd_chariov_t *out) {
         char *sasl_username;
         struct rd_kafka_sasl_scram_state *state = rktrans->rktrans_sasl.state;
-        const rd_kafka_conf_t *conf = &rktrans->rktrans_rkb->rkb_rk->rk_conf;
+        rd_kafka_conf_t *conf = &rktrans->rktrans_rkb->rkb_rk->rk_conf;
 
         rd_kafka_sasl_scram_generate_nonce(&state->cnonce);
 
+        mtx_lock(&conf->sasl.lock);
         sasl_username = rd_kafka_sasl_safe_string(conf->sasl.username);
+        mtx_unlock(&conf->sasl.lock);
 
         out->size =
             strlen("n,,n=,r=") + strlen(sasl_username) + state->cnonce.size;
@@ -842,8 +850,13 @@ static int rd_kafka_sasl_scram_conf_validate(rd_kafka_t *rk,
                                              char *errstr,
                                              size_t errstr_size) {
         const char *mech = rk->rk_conf.sasl.mechanisms;
+        rd_bool_t both_set;
+
+        mtx_lock(&rk->rk_conf.sasl.lock);
+        both_set = rk->rk_conf.sasl.username && rk->rk_conf.sasl.password;
+        mtx_unlock(&rk->rk_conf.sasl.lock);
 
-        if (!rk->rk_conf.sasl.username || !rk->rk_conf.sasl.password) {
+        if (!both_set) {
                 rd_snprintf(errstr, errstr_size,
                             "sasl.username and sasl.password must be set");
                 return -1;
diff --git a/tests/0135-sasl_credentials.cpp b/tests/0135-sasl_credentials.cpp
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
diff --git a/tests/test.c b/tests/test.c
diff --git a/win32/tests/tests.vcxproj b/win32/tests/tests.vcxproj
