From b5f60ddbe5e488390da5f2a2cfeb4ff73b7f2461 Mon Sep 17 00:00:00 2001 From: Pranav Rathi <4427674+pranavrth@users.noreply.github.com> Date: Tue, 7 Jul 2026 09:06:00 +0530 Subject: [PATCH 1/2] Fix IPv6 broker nodename resolution for compressed addresses A metadata-advertised IPv6 host ending in "::" was concatenated as "host:::port" and failed name resolution. Bracket IPv6 literals in rd_kafka_mk_nodename() and make the SASL hostname extraction bracket-aware. Adds broker/sasl unit tests, mock test 0191, and a manual SASL-over-[::1] repro script. --- src/rdkafka_broker.c | 64 +++++++++++++- src/rdkafka_sasl.c | 76 +++++++++++++++-- src/rdunittest.c | 2 + tests/0191-ipv6_nodename_mock.c | 144 ++++++++++++++++++++++++++++++++ tests/CMakeLists.txt | 1 + tests/manual-ipv6-sasl-repro.sh | 102 ++++++++++++++++++++++ tests/test.c | 2 + 7 files changed, 385 insertions(+), 6 deletions(-) create mode 100644 tests/0191-ipv6_nodename_mock.c create mode 100755 tests/manual-ipv6-sasl-repro.sh diff --git a/src/rdkafka_broker.c b/src/rdkafka_broker.c index d8e5c48e81..2229f34658 100644 --- a/src/rdkafka_broker.c +++ b/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; } diff --git a/src/rdkafka_sasl.c b/src/rdkafka_sasl.c index aab16f6d50..006f0ea229 100644 --- a/src/rdkafka_sasl.c +++ b/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; +} diff --git a/src/rdunittest.c b/src/rdunittest.c index fe0e2d84b1..140fdbedab 100644 --- a/src/rdunittest.c +++ b/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}, diff --git a/tests/0191-ipv6_nodename_mock.c b/tests/0191-ipv6_nodename_mock.c new file mode 100644 index 0000000000..f533625346 --- /dev/null +++ b/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: "[]:". */ + 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; +} diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 0f6a0650fe..800bd1e41b 100644 --- a/tests/CMakeLists.txt +++ b/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 diff --git a/tests/manual-ipv6-sasl-repro.sh b/tests/manual-ipv6-sasl-repro.sh new file mode 100755 index 0000000000..4815be92d5 --- /dev/null +++ b/tests/manual-ipv6-sasl-repro.sh @@ -0,0 +1,102 @@ +#!/bin/bash +# +# Manual, opt-in end-to-end check for the IPv6 broker-nodename fix over SASL. +# +# NOT part of the automated suite (it needs Docker + host networking and is a +# .sh, so the test-runner's "[08]*-*.c" glob never picks it up). The automated +# coverage lives in the `broker` / `sasl` unit tests and mock test 0191. +# +# What it proves, against a real Kafka broker listening on the IPv6 loopback +# "[::1]:9092" with SASL_PLAINTEXT/PLAIN: +# 1. Regression: SASL authentication + produce/consume still work with the +# refactored rd_kafka_sasl_client_new(). +# 2. IPv6 fix: the SASL client hostname is extracted as the bare "::1" from +# the bracketed nodename "[::1]:9092" (the pre-fix strchr() code truncated +# it to "["). +# +# A/B the fix: run once with a lib built from this branch (expect PASS), then +# rebuild from origin/master and run again (expect the SASL hostname to be "[" +# and this script to FAIL on that assertion). +# +# Usage: tests/manual-ipv6-sasl-repro.sh +# Env: LIBDIR (default: ../src) librdkafka .so to link the example to +# IMAGE (default: apache/kafka:4.0.0) +# ADDR (default: [::1]) IPv6 host the broker binds/advertises +# +set -u + +LIBDIR="${LIBDIR:-$(cd "$(dirname "$0")/../src" && pwd)}" +EXDIR="$(cd "$(dirname "$0")/../examples" && pwd)" +IMAGE="${IMAGE:-apache/kafka:4.0.0}" +ADDR="${ADDR:-[::1]}" +CONTAINER="kafka-sasl-ipv6-repro" +TOPIC="ipv6sasl-$$" +# Keep the payload short (<=16 bytes) so it fits on a single line of the +# example client's consumer hexdump, keeping the round-trip grep simple. +PAYLOAD="ipv6ok$$" +USER=user +PASS=userpw +CLIENT="$EXDIR/rdkafka_example" +COMMON=(-X security.protocol=SASL_PLAINTEXT -X sasl.mechanism=PLAIN \ + -X sasl.username=$USER -X sasl.password=$PASS) + +fail() { echo "FAIL: $*" >&2; exit 1; } +cleanup() { docker rm -f "$CONTAINER" >/dev/null 2>&1; } +trap cleanup EXIT + +command -v docker >/dev/null || fail "docker not found" +[ -x "$CLIENT" ] || fail "example client not built: run 'make -C src && make -C examples rdkafka_example'" + +echo "## Starting SASL broker on SASL_PLAINTEXT://$ADDR:9092 ($IMAGE)" +cleanup +docker run -d --name "$CONTAINER" --network host \ + -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller \ + -e KAFKA_LISTENERS="SASL://$ADDR:9092,CONTROLLER://$ADDR:9093" \ + -e KAFKA_ADVERTISED_LISTENERS="SASL://$ADDR:9092" \ + -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \ + -e KAFKA_INTER_BROKER_LISTENER_NAME=SASL \ + -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP='SASL:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT' \ + -e KAFKA_CONTROLLER_QUORUM_VOTERS="1@$ADDR:9093" \ + -e KAFKA_SASL_ENABLED_MECHANISMS=PLAIN \ + -e KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN \ + -e KAFKA_LISTENER_NAME_SASL_PLAIN_SASL_JAAS_CONFIG="org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin-secret\" user_admin=\"admin-secret\" user_$USER=\"$PASS\";" \ + -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \ + -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \ + -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \ + -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=true \ + "$IMAGE" >/dev/null || fail "failed to start broker container" + +echo "## Waiting for broker to accept SASL produce" +plog=$(mktemp) +ready=0 +for i in $(seq 1 30); do + if echo "$PAYLOAD" | LD_LIBRARY_PATH="$LIBDIR" timeout 20 \ + "$CLIENT" -P -b "$ADDR:9092" -t "$TOPIC" -p 0 "${COMMON[@]}" \ + -X debug=security,broker >"$plog" 2>&1; then + ready=1; break + fi + sleep 2 +done +[ "$ready" = 1 ] || { docker logs "$CONTAINER" 2>&1 | tail -20; fail "broker never accepted a SASL produce"; } + +echo "## Assertion 1 (regression): SASL authentication reached UP" +grep -q "AUTH_REQ -> UP" "$plog" || fail "SASL auth did not reach UP" + +echo "## Assertion 2 (IPv6 fix): SASL hostname is the bare '::1', not '['" +grep "Initializing SASL client" "$plog" | grep -q "hostname ::1," \ + || { grep "Initializing SASL client" "$plog" >&2; \ + fail "SASL hostname was not extracted as '::1' (pre-fix bug: 'hostname [')"; } +! grep "Initializing SASL client" "$plog" | grep -q "hostname \[" \ + || fail "SASL hostname was mangled to '[' (pre-fix strchr bug present)" +! grep -q ":::" "$plog" || fail "malformed ':::' nodename present (nodename bracket fix missing)" + +echo "## Assertion 3 (regression): message round-trips over SASL" +clog=$(mktemp) +LD_LIBRARY_PATH="$LIBDIR" timeout 20 \ + "$CLIENT" -C -b "$ADDR:9092" -t "$TOPIC" -p 0 -o beginning -e "${COMMON[@]}" \ + >"$clog" 2>&1 +grep -aq "$PAYLOAD" "$clog" || { cat "$clog" >&2; fail "produced message not consumed back over SASL"; } + +echo +echo "PASS: SASL auth + produce/consume over [::1] OK; SASL hostname correctly '::1'" +rm -f "$plog" "$clog" diff --git a/tests/test.c b/tests/test.c index 664376afda..0bf29ab82e 100644 --- a/tests/test.c +++ b/tests/test.c @@ -314,6 +314,7 @@ _TEST_DECL(0184_share_consumer_topic_recreate_local); _TEST_DECL(0185_share_consumer_max_poll_interval); _TEST_DECL(0186_share_consumer_fatal_error); _TEST_DECL(0190_share_consumer_telemetry); +_TEST_DECL(0191_ipv6_nodename_mock); /* Manual tests */ _TEST_DECL(8000_idle); @@ -612,6 +613,7 @@ struct test tests[] = { _TEST(0190_share_consumer_telemetry, TEST_F_MANUAL, TEST_BRKVER(4, 2, 0, 0)), + _TEST(0191_ipv6_nodename_mock, TEST_F_LOCAL), /* Manual tests */ _TEST(8000_idle, TEST_F_MANUAL), From 6afd25e6f4a27a7c887aa86757acd79b1b6d0084 Mon Sep 17 00:00:00 2001 From: Pranav Rathi <4427674+pranavrth@users.noreply.github.com> Date: Tue, 28 Jul 2026 19:21:21 +0530 Subject: [PATCH 2/2] Register 0191 test in the Windows MSVC project The MSVC build uses win32/tests/tests.vcxproj's explicit source list, not the Makefile glob or CMake, so the new test file must be listed there too or test.c fails to link (unresolved main_0191_...). --- win32/tests/tests.vcxproj | 1 + 1 file changed, 1 insertion(+) diff --git a/win32/tests/tests.vcxproj b/win32/tests/tests.vcxproj index ddff9d7ebc..34f37536cb 100644 --- a/win32/tests/tests.vcxproj +++ b/win32/tests/tests.vcxproj @@ -259,6 +259,7 @@ +