Fix IPv6 broker nodename resolution for compressed addresses

src/rdkafka_broker.c

@@ -142,7 +142,14 @@ static void rd_kafka_mk_nodename(char *dest,
                                  size_t dsize,
                                  const char *name,
                                  uint16_t port) {
-        rd_snprintf(dest, dsize, "%s:%hu", name, port);
+        /* An IPv6 literal must be enclosed in brackets so the trailing
+         * ":port" is not mistaken for part of the address. Only a bare
+         * (unbracketed) literal needs wrapping; a hostname or IPv4 address
+         * never contains a ':'. */
+        if (strchr(name, ':') && *name != '[')
+                rd_snprintf(dest, dsize, "[%s]:%hu", name, port);
+        else
+                rd_snprintf(dest, dsize, "%s:%hu", name, port);
 }
 
 /**
@@ -6719,6 +6726,60 @@ static int rd_ut_ApiVersion_at_least(void) {
         RD_UT_PASS();
 }
 
+/**
+ * @brief Unittest for broker nodename construction.
+ *
+ * A nodename produced by rd_kafka_mk_nodename() must be split back into the
+ * original host and port by rd_addrinfo_prepare() (the parsing that feeds
+ * getaddrinfo()). This exercises IPv6 literals, in particular compressed
+ * forms ending in "::", which otherwise concatenate into an unresolvable
+ * ":::port" and fail name resolution.
+ */
+static int rd_ut_mk_nodename(void) {
+        static const struct {
+                const char *host;
+                uint16_t port;
+        } hosts[] = {
+            {"broker.example.com", 9092},
+            {"192.0.2.1", 9092},
+            {"2600:1f18:4dcf:654c:46fc:0:0:1", 9092}, /* full IPv6 */
+            {"2600:1f18:4dcf:654c:46fc::", 9092},     /* compressed tail */
+            {"fe80::", 9092},                         /* compressed */
+            {"::1", 9092},                            /* IPv6 loopback */
+            {NULL, 0},
+        };
+        int i;
+        char nodename[256];
+        char expected_port[16];
+        char *node, *svc;
+        const char *errstr;
+
+        for (i = 0; hosts[i].host; i++) {
+                rd_kafka_mk_nodename(nodename, sizeof(nodename), hosts[i].host,
+                                     hosts[i].port);
+
+                errstr = rd_addrinfo_prepare(nodename, &node, &svc);
+                RD_UT_ASSERT(!errstr,
+                             "host '%s' -> nodename '%s': "
+                             "rd_addrinfo_prepare failed: %s",
+                             hosts[i].host, nodename, errstr);
+
+                RD_UT_ASSERT(!strcmp(node, hosts[i].host),
+                             "host '%s' -> nodename '%s': parsed host '%s' "
+                             "does not match expected '%s'",
+                             hosts[i].host, nodename, node, hosts[i].host);
+
+                rd_snprintf(expected_port, sizeof(expected_port), "%hu",
+                            hosts[i].port);
+                RD_UT_ASSERT(!strcmp(svc, expected_port),
+                             "host '%s' -> nodename '%s': parsed port '%s' "
+                             "does not match expected '%s'",
+                             hosts[i].host, nodename, svc, expected_port);
+        }
+
+        RD_UT_PASS();
+}
+
 /**
  * @name Unit tests
  * @{
@@ -6729,6 +6790,7 @@ int unittest_broker(void) {
 
         fails += rd_ut_reconnect_backoff();
         fails += rd_ut_ApiVersion_at_least();
+        fails += rd_ut_mk_nodename();
 
         return fails;
 }

src/rdkafka_sasl.c

@@ -35,6 +35,7 @@
 #include "rdkafka_sasl_int.h"
 #include "rdkafka_request.h"
 #include "rdkafka_queue.h"
+#include "rdunittest.h"
 
 /**
  * @brief Send SASL auth data using legacy directly on socket framing.
@@ -230,13 +231,39 @@ void rd_kafka_sasl_close(rd_kafka_transport_t *rktrans) {
  *
  * Locality: broker thread
  */
+/**
+ * @brief Extract the bare hostname from a broker nodename for use as the SASL
+ *        hostname, stripping the ":port" suffix and any enclosing IPv6
+ *        brackets (e.g. "[2600::1]:9092" -> "2600::1").
+ */
+static void rd_kafka_sasl_nodename_to_hostname(const char *nodename,
+                                               char *dest,
+                                               size_t dsize) {
+        char *t;
+
+        rd_strlcpy(dest, nodename, dsize);
+
+        /* Strip the ":port" suffix. Use the last ':' so an IPv6 literal such
+         * as "[2600::1]:9092" is not truncated at a ':' within the address. */
+        if ((t = strrchr(dest, ':')))
+                *t = '\0';
+
+        /* Strip the enclosing brackets from an IPv6 literal, leaving the bare
+         * address: "[2600::1]" -> "2600::1". */
+        if (*dest == '[') {
+                memmove(dest, dest + 1, strlen(dest));
+                if ((t = strrchr(dest, ']')))
+                        *t = '\0';
+        }
+}
+
 int rd_kafka_sasl_client_new(rd_kafka_transport_t *rktrans,
                              char *errstr,
                              size_t errstr_size) {
         int r;
         rd_kafka_broker_t *rkb = rktrans->rktrans_rkb;
         rd_kafka_t *rk         = rkb->rkb_rk;
-        char *hostname, *t;
+        char hostname[RD_KAFKA_NODENAME_SIZE];
         const struct rd_kafka_sasl_provider *provider =
             rk->rk_conf.sasl.provider;
 
@@ -263,12 +290,10 @@ int rd_kafka_sasl_client_new(rd_kafka_transport_t *rktrans,
         }
 
         rd_kafka_broker_lock(rktrans->rktrans_rkb);
-        rd_strdupa(&hostname, rktrans->rktrans_rkb->rkb_nodename);
+        rd_kafka_sasl_nodename_to_hostname(rktrans->rktrans_rkb->rkb_nodename,
+                                           hostname, sizeof(hostname));
         rd_kafka_broker_unlock(rktrans->rktrans_rkb);
 
-        if ((t = strchr(hostname, ':')))
-                *t = '\0'; /* remove ":port" */
-
         rd_rkb_dbg(rkb, SECURITY, "SASL",
                    "Initializing SASL client: service name %s, "
                    "hostname %s, mechanisms %s, provider %s",
@@ -587,3 +612,44 @@ rd_kafka_error_t *rd_kafka_sasl_set_credentials(rd_kafka_t *rk,
 
         return NULL;
 }
+
+
+/**
+ * @brief Unittest for SASL hostname extraction from a broker nodename.
+ *
+ * The nodename carries a ":port" suffix, and IPv6 literals are bracketed
+ * ("[2600::1]:9092"); the SASL hostname must be the bare address.
+ */
+static int unittest_sasl_nodename_to_hostname(void) {
+        static const struct {
+                const char *nodename;
+                const char *exp;
+        } tests[] = {
+            {"broker.example.com:9092", "broker.example.com"},
+            {"192.0.2.1:9092", "192.0.2.1"},
+            {"[2600:1f18:4dcf:654c:46fc::]:9092", "2600:1f18:4dcf:654c:46fc::"},
+            {"[fe80::]:9092", "fe80::"},
+            {"[::1]:9092", "::1"},
+            {NULL, NULL},
+        };
+        int i;
+        char hostname[RD_KAFKA_NODENAME_SIZE];
+
+        for (i = 0; tests[i].nodename; i++) {
+                rd_kafka_sasl_nodename_to_hostname(tests[i].nodename, hostname,
+                                                   sizeof(hostname));
+                RD_UT_ASSERT(!strcmp(hostname, tests[i].exp),
+                             "nodename '%s': expected hostname '%s', got '%s'",
+                             tests[i].nodename, tests[i].exp, hostname);
+        }
+
+        RD_UT_PASS();
+}
+
+int unittest_sasl(void) {
+        int fails = 0;
+
+        fails += unittest_sasl_nodename_to_hostname();
+
+        return fails;
+}

src/rdunittest.c

@@ -437,6 +437,7 @@ extern int unittest_sasl_oauthbearer_oidc_assertion(void);
 extern int unittest_admin(void);
 extern int unittest_telemetry(void);
 extern int unittest_telemetry_decode(void);
+extern int unittest_sasl(void);
 #if WITH_SSL
 extern int unittest_ssl(void);
 #endif
@@ -465,6 +466,7 @@ int rd_unittest(void) {
 #endif
             {"conf", unittest_conf},
             {"broker", unittest_broker},
+            {"sasl", unittest_sasl},
             {"request", unittest_request},
 #if WITH_SASL_OAUTHBEARER
             {"sasl_oauthbearer", unittest_sasl_oauthbearer},

tests/0191-ipv6_nodename_mock.c

@@ -0,0 +1,144 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2024, Confluent Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "test.h"
+
+/**
+ * @name Verify that an IPv6 address advertised in a Metadata response is
+ *       turned into a resolvable broker nodename.
+ *
+ * A broker that advertises a compressed IPv6 literal (one ending in "::")
+ * used to produce a nodename such as "2600:...:46fc:::9092" — the trailing
+ * "::" of the address concatenated directly against the ":port", giving a
+ * ":::" that name resolution rejects with "Name or service not known".
+ * The literal must instead be bracketed: "[2600:...:46fc::]:9092".
+ */
+
+/* Compressed IPv6 literal ending in "::", mirroring the customer report. */
+static const char *ipv6_host = "2600:1f18:4dcf:654c:46fc::";
+
+static rd_bool_t nodename_bracketed;
+static rd_bool_t nodename_malformed;
+
+/**
+ * @brief Inspect broker log lines for the nodename derived from the
+ *        advertised IPv6 address.
+ */
+static void ipv6_nodename_mock_log_cb(const rd_kafka_t *rk,
+                                      int level,
+                                      const char *fac,
+                                      const char *buf) {
+        /* The bug manifests as a ":::" run wherever the mangled nodename is
+         * printed (nodename change or a "Failed to resolve" failure). */
+        if (strstr(buf, ":::"))
+                nodename_malformed = rd_true;
+
+        /* The fixed nodename brackets the literal: "[<ipv6>]:<port>". */
+        if (strstr(buf, "Nodename changed") && strstr(buf, "[") &&
+            strstr(buf, ipv6_host))
+                nodename_bracketed = rd_true;
+}
+
+/**
+ * @brief Treat the broker connection errors caused by the (deliberately
+ *        unreachable) advertised address as non-fatal so the test can
+ *        inspect the constructed nodename.
+ */
+static int ipv6_nodename_is_fatal_cb(rd_kafka_t *rk,
+                                     rd_kafka_resp_err_t err,
+                                     const char *reason) {
+        if (err == RD_KAFKA_RESP_ERR__RESOLVE ||
+            err == RD_KAFKA_RESP_ERR__TRANSPORT ||
+            err == RD_KAFKA_RESP_ERR__ALL_BROKERS_DOWN) {
+                TEST_SAY("Ignoring expected error: %s: %s\n",
+                         rd_kafka_err2name(err), reason);
+                return 0;
+        }
+        return 1;
+}
+
+int main_0191_ipv6_nodename_mock(int argc, char **argv) {
+        rd_kafka_mock_cluster_t *cluster;
+        const char *bootstraps;
+        rd_kafka_t *rk;
+        rd_kafka_conf_t *conf;
+        const rd_kafka_metadata_t *md;
+        test_conf_log_interceptor_t *log_interceptor;
+        const char *debug_contexts[2] = {"broker", NULL};
+        int i;
+
+        if (test_needs_auth()) {
+                TEST_SKIP("Mock cluster does not support SSL/SASL\n");
+                return 0;
+        }
+
+        cluster = test_mock_cluster_new(1, &bootstraps);
+
+        test_conf_init(&conf, NULL, tmout_multip(10));
+        test_conf_set(conf, "bootstrap.servers", bootstraps);
+        log_interceptor = test_conf_set_log_interceptor(
+            conf, ipv6_nodename_mock_log_cb, debug_contexts);
+
+        test_curr->is_fatal_cb = ipv6_nodename_is_fatal_cb;
+
+        rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
+
+        TEST_SAY("Initial metadata request (learns 127.0.0.1 nodename)\n");
+        if (!rd_kafka_metadata(rk, 0, NULL, &md, tmout_multip(5000)))
+                rd_kafka_metadata_destroy(md);
+
+        TEST_SAY("Advertising IPv6 broker host %s\n", ipv6_host);
+        rd_kafka_mock_broker_set_host_port(cluster, 1, ipv6_host, 9092);
+
+        TEST_SAY("Metadata request that learns the IPv6 nodename\n");
+        /* The nodename change is applied when this Metadata response arrives
+         * over the still-open bootstrap connection; the request itself may
+         * then time out as the broker becomes unreachable, so don't wait long
+         * for it. */
+        if (!rd_kafka_metadata(rk, 0, NULL, &md, tmout_multip(1000)))
+                rd_kafka_metadata_destroy(md);
+
+        TEST_SAY("Waiting for the IPv6 nodename to be constructed\n");
+        for (i = 0; i < 50 && !nodename_bracketed && !nodename_malformed; i++)
+                rd_kafka_poll(rk, 100);
+
+        TEST_ASSERT(!nodename_malformed,
+                    "IPv6 nodename was built with a malformed \":::\" run "
+                    "(host and port concatenated without brackets)");
+        TEST_ASSERT(nodename_bracketed,
+                    "expected the IPv6 literal to be bracketed as "
+                    "\"[%s]:port\" in the broker nodename",
+                    ipv6_host);
+
+        TEST_SAY("IPv6 nodename correctly bracketed\n");
+        rd_kafka_destroy(rk);
+        test_mock_cluster_destroy(cluster);
+        rd_free(log_interceptor);
+
+        return 0;
+}

tests/CMakeLists.txt

@@ -166,6 +166,7 @@ set(
     0185-share_consumer_max_poll_interval.c
     0186-share_consumer_fatal_error.c
     0190-share_consumer_telemetry.c
+    0191-ipv6_nodename_mock.c
     8000-idle.cpp
     8001-fetch_from_follower_mock_manual.c
     test.c