FIX: Improve PRNG seeding on Windows to ensure uniqueness of generated numbers (#5265)

emasab · web-flow · commit 59b2f66b95ce · 2025-12-19T20:49:01.000+01:00
Fix for the pseudo-random seed on Windows. The function `rand_r` isn't present on Windows and the global seed wasn't based on the current microseconds and thread id. Also it wasn't called on every thread as required on this platform but only once per process. The fix allows on this platform the uniqueness of client side member id generation in next-generation consumer group protocol.
Happening since 1.x

* Multiple platforms secure random generation
* Run KIP-848 tests on MinGW-w64
* Changes to srand per thread on Windows.
It's necessary on Windows to avoid using the same jitter values for all clients and avoid a stampede effect on brokers.
* Run macOS local quick tests with KIP-848 as well
* Run KIP-848 local quick tests on arm64 glibc and alpine
* Increase 0153 with a different number of maximum open files
diff --git a/.semaphore/semaphore-integration.yml b/.semaphore/semaphore-integration.yml
@@ -47,15 +47,7 @@ blocks:
       jobs:
         - name: 'Build'
           commands:
-            - ./configure --install-deps --source-deps-only --enable-static --disable-lz4-ext --enable-strip
-            - make -j all examples check
-            - examples/rdkafka_example -X builtin.features
-            - otool -L src/librdkafka.dylib
-            - otool -L src-cpp/librdkafka++.dylib
-            - make -j -C tests build
-            - make -C tests run_local_quick
-            - DESTDIR="$PWD/dest" make install
-            - (cd dest && tar cvzf ../artifacts/librdkafka.tgz .)
+            - packaging/macos/build-release-artifacts.sh
 
 
   - name: 'OSX x64'
@@ -73,15 +65,7 @@ blocks:
       jobs:
         - name: 'Build'
           commands:
-            - ./configure --install-deps --source-deps-only --enable-static --disable-lz4-ext --enable-strip
-            - make -j all examples check
-            - examples/rdkafka_example -X builtin.features
-            - otool -L src/librdkafka.dylib
-            - otool -L src-cpp/librdkafka++.dylib
-            - make -j -C tests build
-            - make -C tests run_local_quick
-            - DESTDIR="$PWD/dest" make install
-            - (cd dest && tar cvzf ../artifacts/librdkafka.tgz .)
+            - packaging/macos/build-release-artifacts.sh
 
   - name: 'Linux Ubuntu amd64: integration tests'
     dependencies: []
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,9 +2,17 @@
 
 librdkafka v2.13.0 is a feature release:
 
-* Strip trailing dot of hostname to fix SSL certificate verification issue (#5253).
 * [KIP-482](https://cwiki.apache.org/confluence/display/KAFKA/KIP-482%3A+The+Kafka+Protocol+should+Support+Optional+Tagged+Fields) Upgrade CreateAcls, DescribeAcls, DeleteAcls to the first version supporting this KIP (#5081).
 * [KIP-482](https://cwiki.apache.org/confluence/display/KAFKA/KIP-482%3A+The+Kafka+Protocol+should+Support+Optional+Tagged+Fields) Upgrade DescribeGroups, DeleteTopics, DeleteRecords, CreatePartitions, DeleteGroups to the first version supporting this KIP (#5083).
+* Strip trailing dot of hostname to fix SSL certificate verification issue (#5253).
+* Fix memory management for interceptors in rd_kafka_conf to prevent
+double-free errors (#5240).
+* Fix for the pseudo-random generator seed on Windows involving as well
+  the uniqueness of the new consumer group protocol member id (#5265).
+* Add secure random generation functionality used for UUID uniqueness
+  and secure salt generation in `rd_kafka_UserScramCredentialUpsertion`
+  using OpenSSL or the POSIX or WIN32 equivalent calls when it
+  isn't available (#5265).
 
 
 ## Fixes
@@ -14,6 +22,19 @@ librdkafka v2.13.0 is a feature release:
 * Issues: #4348.
   Strip trailing dot of hostname to fix SSL certificate verification issue.
   Happening since 1.x (#5253).
+* Issues: #4142.
+  Fix memory management for interceptors in rd_kafka_conf to prevent double-free errors.
+  In case the client instance fails the users needs to destroy the configuration
+  data structure, it was causing a double-free because the interceptors were
+  already freed in the constructor.
+  Happening since 1.x (#5240).
+* Issues: #5263, #3929.
+  Fix for the pseudo-random seed on Windows. The function `rand_r` isn't present
+  on Windows and the global seed wasn't based on the current microseconds and thread
+  id. Also it wasn't called on every thread as required on this platform but
+  only once per process. The fix allows on this platform the uniqueness of client side 
+  member id generation in next-generation consumer group protocol.
+  Happening since 1.x (#5265).
 
 
 
diff --git a/configure.self b/configure.self
@@ -216,6 +216,16 @@ void foo (void) {
    (void)rand_r(&seed);
 }"
 
+    # Check if getentropy() is available
+    mkl_compile_check "getentropy" "HAVE_GETENTROPY" disable CC "" \
+"#define _DEFAULT_SOURCE
+#include <unistd.h>
+#include <sys/random.h>
+int foo (void) {
+   char seed[16];
+   return getentropy((void *)seed, sizeof(seed));
+}"
+
     # Check if strndup() is available (isn't on Solaris 10)
     mkl_compile_check "strndup" "HAVE_STRNDUP" disable CC "" \
 "#include <string.h>
diff --git a/packaging/cmake/config.h.in b/packaging/cmake/config.h.in
@@ -43,6 +43,7 @@
 #cmakedefine01 HAVE_REGEX
 #cmakedefine01 HAVE_STRNDUP
 #cmakedefine01 HAVE_RAND_R
+#cmakedefine01 HAVE_GETENTROPY
 #cmakedefine01 HAVE_PTHREAD_SETNAME_GNU
 #cmakedefine01 HAVE_PTHREAD_SETNAME_DARWIN
 #cmakedefine01 HAVE_PTHREAD_SETNAME_FREEBSD
diff --git a/packaging/cmake/try_compile/getentropy_test.c b/packaging/cmake/try_compile/getentropy_test.c
@@ -0,0 +1,7 @@
+#include <unistd.h>
+#include <sys/random.h>
+
+int main() {
+        char seed[16];
+        return getentropy(seed, sizeof(seed));
+}
diff --git a/packaging/cmake/try_compile/rdkafka_setup.cmake b/packaging/cmake/try_compile/rdkafka_setup.cmake
@@ -16,6 +16,12 @@ try_compile(
     "${TRYCOMPILE_SRC_DIR}/rand_r_test.c"
 )
 
+try_compile(
+    HAVE_GETENTROPY
+    "${CMAKE_CURRENT_BINARY_DIR}/try_compile"
+    "${TRYCOMPILE_SRC_DIR}/getentropy_test.c"
+)
+
 try_compile(
     HAVE_PTHREAD_SETNAME_GNU
     "${CMAKE_CURRENT_BINARY_DIR}/try_compile"
diff --git a/packaging/macos/build-release-artifacts.sh b/packaging/macos/build-release-artifacts.sh
@@ -0,0 +1,26 @@
+./configure --install-deps --source-deps-only --enable-static --disable-lz4-ext --enable-strip
+make -j all examples check
+examples/rdkafka_example -X builtin.features
+otool -L src/librdkafka.dylib
+otool -L src-cpp/librdkafka++.dylib
+make -j -C tests build
+export TEST_CONSUMER_GROUP_PROTOCOL=classic
+make -C tests run_local_quick
+export TEST_CONSUMER_GROUP_PROTOCOL=consumer
+# Skip tests needing special limits
+TESTS_WITH_INCREASED_NLIMIT="0153"
+export TESTS_SKIP="$TESTS_WITH_INCREASED_NLIMIT"
+make -C tests run_local_quick
+# Now run only those tests with different limits
+
+# Tests needing increased number of file descriptors
+PREV_N=$(ulimit -n)
+ulimit -n 2048
+export TESTS_SKIP=""
+export TESTS="$TESTS_WITH_INCREASED_NLIMIT"
+make -C tests run_local_quick
+ulimit -n $PREV_N
+
+
+DESTDIR="$PWD/dest" make install
+(cd dest && tar cvzf ../artifacts/librdkafka.tgz .)
diff --git a/packaging/mingw-w64/run-tests.sh b/packaging/mingw-w64/run-tests.sh
@@ -3,4 +3,7 @@
 set -e
 
 cd tests
-./test-runner.exe -l -Q -p1 0000
+export TEST_CONSUMER_GROUP_PROTOCOL=classic
+./test-runner.exe -l -Q -p1
+export TEST_CONSUMER_GROUP_PROTOCOL=consumer
+./test-runner.exe -l -Q -p1 
diff --git a/packaging/tools/build-release-artifacts.sh b/packaging/tools/build-release-artifacts.sh
@@ -123,6 +123,7 @@ fi
 # Run quick test suite, mark it as CI to avoid time/resource sensitive
 # tests to fail in case the worker is under-powered.
 CI=true make -C tests run_local_quick
+CI=true TEST_CONSUMER_GROUP_PROTOCOL=consumer make -C tests run_local_quick
 
 
 # Install librdkafka and then make a tar ball of the installed files.
diff --git a/src/rd.h b/src/rd.h
@@ -44,6 +44,11 @@
 #ifndef _POSIX_C_SOURCE
 #define _POSIX_C_SOURCE 200809L /* for timespec on solaris */
 #endif
+#else
+#ifndef _CRT_RAND_S
+#define _CRT_RAND_S  /* for rand_s() on MSVC. It needs to be defined before    \
+                      * including <stdlib.h>. */
+#endif
 #endif
 
 #include <stdio.h>
diff --git a/src/rdkafka.c b/src/rdkafka.c
@@ -82,8 +82,14 @@
 #endif
 
 
-static once_flag rd_kafka_global_init_once  = ONCE_FLAG_INIT;
-static once_flag rd_kafka_global_srand_once = ONCE_FLAG_INIT;
+static once_flag rd_kafka_global_init_once = ONCE_FLAG_INIT;
+#ifdef _WIN32
+/* On Windows srand needs to be called on each thread. */
+static RD_TLS once_flag rd_kafka_srand_once = ONCE_FLAG_INIT;
+#else
+static once_flag rd_kafka_srand_once = ONCE_FLAG_INIT;
+#endif
+
 
 /**
  * @brief Global counter+lock for all active librdkafka instances
@@ -130,6 +136,13 @@ void rd_kafka_set_thread_name(const char *fmt, ...) {
  */
 static char RD_TLS rd_kafka_thread_sysname[16] = "app";
 
+/**
+ * @brief Seed the PRNG with current microseconds and thread ID.
+ */
+static void rd_kafka_srand(void) {
+        srand(rd_seed());
+}
+
 void rd_kafka_set_thread_sysname(const char *fmt, ...) {
         va_list ap;
 
@@ -141,6 +154,31 @@ void rd_kafka_set_thread_sysname(const char *fmt, ...) {
         thrd_setname(rd_kafka_thread_sysname);
 }
 
+/**
+ * @brief Seed the PRNG for the current thread or for the whole process.
+ *        Depending on the platform implementation of srand() the seed can
+ *        be a thread local or global one. In case it's thread local we
+ *        need to call it on each thread.
+ *
+ * @param rk Client instance.
+ * @param internal_thread If true, seed the PRNG if
+ *                        it's required per-thread.
+ */
+#ifdef _WIN32
+void rd_kafka_thread_srand(rd_kafka_t *rk, rd_bool_t internal_thread) {
+        /* We always call it once per thread in the internal threads and
+         * once per thread in the application threads, if requested. */
+        if (rk->rk_conf.enable_random_seed || internal_thread)
+                call_once(&rd_kafka_srand_once, rd_kafka_srand);
+}
+#else
+void rd_kafka_thread_srand(rd_kafka_t *rk, rd_bool_t internal_thread) {
+        /* We call it only in the app threads, only once, if requested. */
+        if (rk->rk_conf.enable_random_seed && !internal_thread)
+                call_once(&rd_kafka_srand_once, rd_kafka_srand);
+}
+#endif
+
 static void rd_kafka_global_init0(void) {
         cJSON_Hooks json_hooks = {.malloc_fn = rd_malloc, .free_fn = rd_free};
 
@@ -171,18 +209,6 @@ void rd_kafka_global_init(void) {
 }
 
 
-/**
- * @brief Seed the PRNG with current_time.milliseconds
- */
-static void rd_kafka_global_srand(void) {
-        struct timeval tv;
-
-        rd_gettimeofday(&tv, NULL);
-
-        srand((unsigned int)(tv.tv_usec / 1000));
-}
-
-
 /**
  * @returns the current number of active librdkafka instances
  */
@@ -2218,6 +2244,7 @@ static int rd_kafka_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("main");
         rd_kafka_set_thread_sysname("rdk:main");
+        rd_kafka_thread_srand(rk, rd_true /* we're in an internal thread */);
 
         rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_MAIN);
 
@@ -2367,10 +2394,7 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
                                 * freed from rd_kafka_destroy_internal()
                                 * as the rk itself is destroyed. */
 
-        /* Seed PRNG, don't bother about HAVE_RAND_R, since it is pretty cheap.
-         */
-        if (rk->rk_conf.enable_random_seed)
-                call_once(&rd_kafka_global_srand_once, rd_kafka_global_srand);
+        rd_kafka_thread_srand(rk, rd_false /* we're on an app thread */);
 
         /* Call on_new() interceptors */
         rd_kafka_interceptors_on_new(rk, &rk->rk_conf);
@@ -5396,18 +5420,12 @@ rd_kafka_Uuid_t *rd_kafka_Uuid_copy(const rd_kafka_Uuid_t *uuid) {
  * @remark Must be freed after use using rd_kafka_Uuid_destroy().
  */
 rd_kafka_Uuid_t rd_kafka_Uuid_random() {
-        int i;
         unsigned char rand_values_bytes[16] = {0};
         uint64_t *rand_values_uint64        = (uint64_t *)rand_values_bytes;
-        unsigned char *rand_values_app;
-        rd_kafka_Uuid_t ret = RD_KAFKA_UUID_ZERO;
-        for (i = 0; i < 16; i += 2) {
-                uint16_t rand_uint16 = (uint16_t)rd_jitter(0, INT16_MAX - 1);
-                /* No need to convert endianess here because it's still only
-                 * a random value. */
-                rand_values_app = (unsigned char *)&rand_uint16;
-                rand_values_bytes[i] |= rand_values_app[0];
-                rand_values_bytes[i + 1] |= rand_values_app[1];
+        rd_kafka_Uuid_t ret                 = RD_KAFKA_UUID_ZERO;
+        if (!rd_rand_bytes(rand_values_bytes, sizeof(rand_values_bytes))) {
+                rd_assert(!*"BUG: a random UUID cannot be requested without "
+                           "a reliable entropy source");
         }
 
         rand_values_bytes[6] &= 0x0f; /* clear version */
diff --git a/src/rdkafka_admin.c b/src/rdkafka_admin.c
@@ -6042,9 +6042,9 @@ rd_kafka_UserScramCredentialUpsertion_new(const char *username,
                 alteration->alteration.upsertion.salt =
                     rd_kafkap_bytes_new(salt, salt_size);
         } else {
-#if WITH_SSL && OPENSSL_VERSION_NUMBER >= 0x10101000L
+#if HAVE_SECURE_RAND_BYTES
                 unsigned char random_salt[64];
-                if (RAND_priv_bytes(random_salt, sizeof(random_salt)) == 1) {
+                if (rd_rand_bytes(random_salt, sizeof(random_salt))) {
                         alteration->alteration.upsertion.salt =
                             rd_kafkap_bytes_new(random_salt,
                                                 sizeof(random_salt));
diff --git a/src/rdkafka_admin.h b/src/rdkafka_admin.h
@@ -35,10 +35,7 @@
 #include "rdmap.h"
 #include "rdkafka_error.h"
 #include "rdkafka_confval.h"
-
-#if WITH_SSL && OPENSSL_VERSION_NUMBER >= 0x10101000L
-#include <openssl/rand.h>
-#endif
+#include "rdrand.h"
 
 #if WITH_SSL
 typedef struct rd_kafka_broker_s rd_kafka_broker_t;
diff --git a/src/rdkafka_background.c b/src/rdkafka_background.c
@@ -111,6 +111,7 @@ int rd_kafka_background_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("background");
         rd_kafka_set_thread_sysname("rdk:bg");
+        rd_kafka_thread_srand(rk, rd_true /* we're in an internal thread */);
 
         rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_BACKGROUND);
 
diff --git a/src/rdkafka_broker.c b/src/rdkafka_broker.c
@@ -4509,6 +4509,7 @@ static int rd_kafka_broker_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("%s", rkb->rkb_name);
         rd_kafka_set_thread_sysname("rdk:broker%" PRId32, rkb->rkb_nodeid);
+        rd_kafka_thread_srand(rk, rd_true /* we're in an internal thread */);
 
         rd_kafka_interceptors_on_thread_start(rk, RD_KAFKA_THREAD_BROKER);
 
diff --git a/src/rdkafka_conf.c b/src/rdkafka_conf.c
@@ -4229,6 +4229,11 @@ const char *rd_kafka_conf_finalize(rd_kafka_type_t cltype,
                                 return "`max.poll.interval.ms`must be >= "
                                        "`session.timeout.ms`";
                 } else {
+#if !HAVE_SECURE_RAND_BYTES
+                        return "`group.protocol=consumer` requires "
+                               "librdkafka to be built with secure random "
+                               "number generator support";
+#endif
 
                         if (rd_kafka_conf_is_modified(conf,
                                                       "session.timeout.ms")) {
diff --git a/src/rdkafka_int.h b/src/rdkafka_int.h
@@ -1122,6 +1122,7 @@ extern char RD_TLS rd_kafka_thread_name[64];
 
 void rd_kafka_set_thread_name(const char *fmt, ...) RD_FORMAT(printf, 1, 2);
 void rd_kafka_set_thread_sysname(const char *fmt, ...) RD_FORMAT(printf, 1, 2);
+void rd_kafka_thread_srand(rd_kafka_t *rk, rd_bool_t internal_thread);
 
 int rd_kafka_path_is_dir(const char *path);
 rd_bool_t rd_kafka_dir_is_empty(const char *path);
diff --git a/src/rdkafka_mock.c b/src/rdkafka_mock.c
@@ -1490,6 +1490,8 @@ static int rd_kafka_mock_cluster_thread_main(void *arg) {
 
         rd_kafka_set_thread_name("mock");
         rd_kafka_set_thread_sysname("rdk:mock");
+        rd_kafka_thread_srand(mcluster->rk,
+                              rd_true /* we're in an internal thread */);
         rd_kafka_interceptors_on_thread_start(mcluster->rk,
                                               RD_KAFKA_THREAD_BACKGROUND);
         rd_atomic32_add(&rd_kafka_thread_cnt_curr, 1);
diff --git a/src/rdkafka_ssl.c b/src/rdkafka_ssl.c
@@ -2045,14 +2045,7 @@ rd_kafka_transport_ssl_lock_cb(int mode, int i, const char *file, int line) {
 #endif
 
 static RD_UNUSED unsigned long rd_kafka_transport_ssl_threadid_cb(void) {
-#ifdef _WIN32
-        /* Windows makes a distinction between thread handle
-         * and thread id, which means we can't use the
-         * thrd_current() API that returns the handle. */
-        return (unsigned long)GetCurrentThreadId();
-#else
-        return (unsigned long)(intptr_t)thrd_current();
-#endif
+        return thrd_current_id();
 }
 
 #ifdef HAVE_OPENSSL_CRYPTO_THREADID_SET_CALLBACK
diff --git a/src/rdrand.c b/src/rdrand.c
diff --git a/src/rdrand.h b/src/rdrand.h
diff --git a/src/tinycthread_extra.c b/src/tinycthread_extra.c
diff --git a/src/tinycthread_extra.h b/src/tinycthread_extra.h
diff --git a/tests/0148-offset_fetch_commit_error_mock.c b/tests/0148-offset_fetch_commit_error_mock.c
diff --git a/tests/test.c b/tests/test.c
diff --git a/win32/msbuild.ps1 b/win32/msbuild.ps1
