From 1e9997d9be4eafc4bcbf55c1df2fdf0181607934 Mon Sep 17 00:00:00 2001 From: sujay-d07 Date: Tue, 19 May 2026 01:06:09 +0530 Subject: [PATCH] rgw/kafka: add support for GSSAPI mechanism for Kafka Bucket Notification Endpoint Signed-off-by: sujay-d07 --- doc/radosgw/notifications.rst | 20 +++ src/common/options/rgw.yaml.in | 11 ++ src/rgw/driver/rados/rgw_pubsub_push.cc | 14 +- src/rgw/rgw_kafka.cc | 176 +++++++++++++++++++----- src/rgw/rgw_kafka.h | 13 +- src/rgw/rgw_rest_pubsub.cc | 3 +- 6 files changed, 199 insertions(+), 38 deletions(-) diff --git a/doc/radosgw/notifications.rst b/doc/radosgw/notifications.rst index de07940ef61..2587501a5f0 100644 --- a/doc/radosgw/notifications.rst +++ b/doc/radosgw/notifications.rst @@ -180,6 +180,10 @@ configuration, we should send smaller batches using: .. confval:: rgw_kafka_max_batch_size +Default Kerberos service name to be used for Kafka SASL/GSSAPI: + +.. confval:: rgw_kafka_sasl_kerberos_service_name + Bucket Notification REST API ---------------------------- @@ -239,6 +243,9 @@ and must be between 1 and 256 characters long. To relax these requirements, use: [&Attributes.entry.19.key=ssl-certificate-location&Attributes.entry.19.value=] [&Attributes.entry.20.key=ssl-key-location&Attributes.entry.20.value=] [&Attributes.entry.21.key=ssl-key-password&Attributes.entry.21.value=] + [&Attributes.entry.22.key=sasl-kerberos-service-name&Attributes.entry.22.value=] + [&Attributes.entry.23.key=sasl-kerberos-principal&Attributes.entry.23.value=] + [&Attributes.entry.24.key=sasl-kerberos-keytab&Attributes.entry.24.value=] Request parameters: @@ -382,6 +389,14 @@ Request parameters: for ``user``/``password``: it should be provided over HTTPS or ``rgw_allow_notification_secrets_in_cleartext`` must be set to "true". + - ``sasl-kerberos-service-name``: Kerberos service name used with + ``GSSAPI`` (per-topic override). If not provided, the global + config value ``rgw_kafka_sasl_kerberos_service_name`` is used. + - ``sasl-kerberos-principal``: Kerberos principal for the RGW client when + using ``GSSAPI``. Optional if a valid ticket is available in the cache. + - ``sasl-kerberos-keytab``: Path to the keytab to use with ``GSSAPI``. + Optional if a valid ticket is available in the cache. + .. note:: - The key-value pair of a specific parameter need not reside in the same @@ -658,12 +673,17 @@ Valid ``AttributeName`` that can be passed: broker before being delivered to their final destinations. - ``kafka-brokers``: Set endpoint with broker(s) as a comma-separated list of ``host`` or ``host:port`` (default port 9092). +- ``user-name``: User name to use when connecting to the Kafka broker. +- ``password``: Password to use when connecting to the Kafka broker. - ``ssl-certificate-location``: Path to a PEM-encoded client certificate for mTLS authentication to the Kafka broker. Must be provided together with ``ssl-key-location``; specifying only one will cause the connection to fail. - ``ssl-key-location``: Path to a PEM-encoded private key corresponding to the client certificate. Must be provided together with ``ssl-certificate-location``. - ``ssl-key-password``: Password for an encrypted private key (optional). +- ``sasl-kerberos-service-name``: Kerberos service name for Kafka SASL/GSSAPI. +- ``sasl-kerberos-principal``: Kerberos principal for the RGW client when using ``GSSAPI``. +- ``sasl-kerberos-keytab``: Path to the keytab to use with ``GSSAPI``. Notifications ~~~~~~~~~~~~~ diff --git a/src/common/options/rgw.yaml.in b/src/common/options/rgw.yaml.in index 3888e2f92b9..921069cfdd0 100644 --- a/src/common/options/rgw.yaml.in +++ b/src/common/options/rgw.yaml.in @@ -4647,6 +4647,17 @@ options: see_also: - rgw_bucket_counters_cache with_legacy: true +- name: rgw_kafka_sasl_kerberos_service_name + type: str + level: advanced + desc: Kerberos service name for Kafka SASL/GSSAPI + long_desc: Sets sasl.kerberos.service.name for the Kafka notification endpoint + when using SASL/GSSAPI. Can be overridden per topic via the + sasl-kerberos-service-name topic attribute. + default: kafka + services: + - rgw + with_legacy: true - name: rgw_kafka_connection_idle type: uint level: advanced diff --git a/src/rgw/driver/rados/rgw_pubsub_push.cc b/src/rgw/driver/rados/rgw_pubsub_push.cc index 3fdf2e9c502..69b9d3ae240 100644 --- a/src/rgw/driver/rados/rgw_pubsub_push.cc +++ b/src/rgw/driver/rados/rgw_pubsub_push.cc @@ -306,12 +306,18 @@ public: ack_level(get_ack_level(args)) { if (!kafka::connect( conn_id, _endpoint, get_bool(args, "use-ssl", false), - get_bool(args, "verify-ssl", true), args.get_optional("ca-location"), - args.get_optional("mechanism"), args.get_optional("user-name"), - args.get_optional("password"), args.get_optional("kafka-brokers"), + get_bool(args, "verify-ssl", true), + args.get_optional("ca-location"), + args.get_optional("mechanism"), + args.get_optional("user-name"), + args.get_optional("password"), + args.get_optional("kafka-brokers"), args.get_optional("ssl-certificate-location"), args.get_optional("ssl-key-location"), - args.get_optional("ssl-key-password"))) { + args.get_optional("ssl-key-password"), + args.get_optional("sasl-kerberos-service-name"), + args.get_optional("sasl-kerberos-principal"), + args.get_optional("sasl-kerberos-keytab"))) { throw configuration_error("Kafka: failed to create connection to: " + _endpoint); } diff --git a/src/rgw/rgw_kafka.cc b/src/rgw/rgw_kafka.cc index c86b8445723..6043c92b4bd 100644 --- a/src/rgw/rgw_kafka.cc +++ b/src/rgw/rgw_kafka.cc @@ -80,7 +80,10 @@ connection_id_t::connection_id_t( bool _verify_ssl, const boost::optional& _ssl_certificate, const boost::optional& _ssl_key, - const boost::optional& _ssl_key_password + const boost::optional& _ssl_key_password, + const boost::optional& _kerberos_service_name, + const boost::optional& _kerberos_principal, + const boost::optional& _kerberos_keytab ) : broker(_broker), user(_user), password(_password), ssl(_ssl), verify_ssl(_verify_ssl) { @@ -100,6 +103,15 @@ connection_id_t::connection_id_t( if (_ssl_key_password.has_value()) { ssl_key_password = _ssl_key_password.get(); } + if (_kerberos_service_name.has_value()) { + kerberos_service_name = _kerberos_service_name.get(); + } + if (_kerberos_principal.has_value()) { + kerberos_principal = _kerberos_principal.get(); + } + if (_kerberos_keytab.has_value()) { + kerberos_keytab = _kerberos_keytab.get(); + } } // equality operator and hasher functor are needed @@ -111,7 +123,10 @@ bool operator==(const connection_id_t& lhs, const connection_id_t& rhs) { lhs.verify_ssl == rhs.verify_ssl && lhs.ssl_certificate == rhs.ssl_certificate && lhs.ssl_key == rhs.ssl_key && - lhs.ssl_key_password == rhs.ssl_key_password; + lhs.ssl_key_password == rhs.ssl_key_password && + lhs.kerberos_service_name == rhs.kerberos_service_name && + lhs.kerberos_principal == rhs.kerberos_principal && + lhs.kerberos_keytab == rhs.kerberos_keytab; } struct connection_id_hasher { @@ -127,12 +142,32 @@ struct connection_id_hasher { boost::hash_combine(h, k.ssl_certificate); boost::hash_combine(h, k.ssl_key); boost::hash_combine(h, k.ssl_key_password); + boost::hash_combine(h, k.kerberos_service_name); + boost::hash_combine(h, k.kerberos_principal); + boost::hash_combine(h, k.kerberos_keytab); return h; } }; std::string to_string(const connection_id_t& id) { - return id.broker + ":" + id.user; + std::ostringstream ss; + ss << id.broker; + if (!id.user.empty()) { + ss << ":" << id.user; + } + if (!id.mechanism.empty()) { + ss << " mechanism=" << id.mechanism; + } + if (!id.kerberos_service_name.empty()) { + ss << " service=" << id.kerberos_service_name; + } + if (!id.kerberos_principal.empty()) { + ss << " principal=" << id.kerberos_principal; + } + if (!id.kerberos_keytab.empty()) { + ss << " keytab=" << id.kerberos_keytab; + } + return ss.str(); } // convert int status to errno - both RGW and librdkafka values @@ -191,6 +226,9 @@ struct connection_t { const boost::optional ssl_certificate; const boost::optional ssl_key; const boost::optional ssl_key_password; + const boost::optional kerberos_service_name; + const boost::optional kerberos_principal; + const boost::optional kerberos_keytab; utime_t timestamp = ceph_clock_now(); // cleanup of all internal connection resource @@ -226,9 +264,13 @@ struct connection_t { const std::string& _user, const std::string& _password, const boost::optional& _mechanism, const boost::optional& _ssl_certificate, const boost::optional& _ssl_key, - const boost::optional& _ssl_key_password) : + const boost::optional& _ssl_key_password, + const boost::optional& _kerberos_service_name, + const boost::optional& _kerberos_principal, + const boost::optional& _kerberos_keytab) : cct(_cct), broker(_broker), use_ssl(_use_ssl), verify_ssl(_verify_ssl), ca_location(_ca_location), user(_user), password(_password), mechanism(_mechanism), - ssl_certificate(_ssl_certificate), ssl_key(_ssl_key), ssl_key_password(_ssl_key_password) {} + ssl_certificate(_ssl_certificate), ssl_key(_ssl_key), ssl_key_password(_ssl_key_password), + kerberos_service_name(_kerberos_service_name), kerberos_principal(_kerberos_principal), kerberos_keytab(_kerberos_keytab) {} // dtor also destroys the internals ~connection_t() { @@ -312,6 +354,8 @@ bool new_producer(connection_t* conn) { char errstr[512] = {0}; + const bool is_gssapi = conn->mechanism && boost::iequals(*conn->mechanism, "GSSAPI"); + // set message timeout // according to documentation, value of zero will expire the message based on retries. // however, testing with librdkafka v1.6.1 did not expire the message in that case. hence, a value of zero is changed to 1ms @@ -336,12 +380,32 @@ bool new_producer(connection_t* conn) { // get list of brokers based on the bootstrap broker if (rd_kafka_conf_set(conf.get(), "bootstrap.servers", conn->broker.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; - if (conn->use_ssl) { + if (is_gssapi) { + const char* security_protocol = conn->use_ssl ? "SASL_SSL" : "SASL_PLAINTEXT"; + if (rd_kafka_conf_set(conf.get(), "security.protocol", security_protocol, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || + rd_kafka_conf_set(conf.get(), "sasl.mechanism", "GSSAPI", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + ldout(conn->cct, 20) << "Kafka connect: configured GSSAPI SASL" << dendl; + + if (conn->kerberos_service_name) { + if (rd_kafka_conf_set(conf.get(), "sasl.kerberos.service.name", conn->kerberos_service_name->c_str(), errstr, + sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + } + if (conn->kerberos_principal) { + if (rd_kafka_conf_set(conf.get(), "sasl.kerberos.principal", conn->kerberos_principal->c_str(), errstr, + sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + } + if (conn->kerberos_keytab) { + if (rd_kafka_conf_set(conf.get(), "sasl.kerberos.keytab", conn->kerberos_keytab->c_str(), errstr, + sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + } else { + ldout(conn->cct, 20) << "Kafka connect: GSSAPI without keytab; using ticket cache" << dendl; + } + } else if (conn->use_ssl) { if (!conn->user.empty()) { // use SSL+SASL if (rd_kafka_conf_set(conf.get(), "security.protocol", "SASL_SSL", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || - rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || - rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || + rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; ldout(conn->cct, 20) << "Kafka connect: successfully configured SSL+SASL security" << dendl; if (conn->mechanism) { @@ -385,19 +449,30 @@ bool new_producer(connection_t* conn) { ldout(conn->cct, 20) << "Kafka connect: successfully configured security" << dendl; } else if (!conn->user.empty()) { - // use SASL+PLAINTEXT - if (rd_kafka_conf_set(conf.get(), "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || - rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || - rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; - ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL_PLAINTEXT" << dendl; + // use SASL+PLAINTEXT + if (rd_kafka_conf_set(conf.get(), "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || + rd_kafka_conf_set(conf.get(), "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK || + rd_kafka_conf_set(conf.get(), "sasl.password", conn->password.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL_PLAINTEXT" << dendl; + + if (conn->mechanism) { + if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", conn->mechanism->c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL mechanism" << dendl; + } else { + if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + ldout(conn->cct, 20) << "Kafka connect: using default SASL mechanism" << dendl; + } + } - if (conn->mechanism) { - if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", conn->mechanism->c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; - ldout(conn->cct, 20) << "Kafka connect: successfully configured SASL mechanism" << dendl; - } else { - if (rd_kafka_conf_set(conf.get(), "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; - ldout(conn->cct, 20) << "Kafka connect: using default SASL mechanism" << dendl; - } + if (is_gssapi && conn->use_ssl) { + if (conn->ca_location) { + if (rd_kafka_conf_set(conf.get(), "ssl.ca.location", conn->ca_location->c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + ldout(conn->cct, 20) << "Kafka connect: successfully configured CA location" << dendl; + } else { + ldout(conn->cct, 20) << "Kafka connect: using default CA location" << dendl; + } + if (rd_kafka_conf_set(conf.get(), "enable.ssl.certificate.verification", conn->verify_ssl ? "true" : "false", errstr, + sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; } // set the global callback for delivery success/fail @@ -664,7 +739,10 @@ public: boost::optional brokers, boost::optional ssl_certificate, boost::optional ssl_key, - boost::optional ssl_key_password) { + boost::optional ssl_key_password, + boost::optional topic_kerberos_service_name, + boost::optional topic_kerberos_principal, + boost::optional topic_kerberos_keytab) { if (stopped) { ldout(cct, 1) << "Kafka connect: manager is stopped" << dendl; return false; @@ -694,20 +772,49 @@ public: password = topic_password.get(); } - // this should be validated by the regex in parse_url() - ceph_assert(user.empty() == password.empty()); + const bool is_gssapi = mechanism.has_value() && boost::iequals(mechanism.get(), "GSSAPI"); - if (!user.empty() && !use_ssl && !g_conf().get_val("rgw_allow_notification_secrets_in_cleartext")) { - ldout(cct, 1) << "Kafka connect: user/password are only allowed over secure connection" << dendl; - return false; - } + if (is_gssapi) { + if (!user.empty() || !password.empty()) { + ldout(cct, 5) << "Kafka connect: user/password provided with GSSAPI; ignoring" << dendl; + } + user.clear(); + password.clear(); + } else { + // this should be validated by the regex in parse_url() + ceph_assert(user.empty() == password.empty()); + if (!user.empty() && !use_ssl && !g_conf().get_val("rgw_allow_notification_secrets_in_cleartext")) { + ldout(cct, 1) << "Kafka connect: user/password are only allowed over secure connection" << dendl; + return false; + } + } + // ssl_certificate and ssl_key must both be provided for mTLS if (ssl_certificate.has_value() != ssl_key.has_value()) { ldout(cct, 1) << "Kafka connect: both ssl_certificate and ssl_key must be provided for mTLS (got only " << (ssl_certificate.has_value() ? "ssl_certificate" : "ssl_key") << ")" << dendl; return false; } + + boost::optional kerberos_service_name; + boost::optional kerberos_principal; + boost::optional kerberos_keytab; + + if (is_gssapi) { + if (topic_kerberos_service_name.has_value()) { + kerberos_service_name = topic_kerberos_service_name; + } else if (!cct->_conf->rgw_kafka_sasl_kerberos_service_name.empty()) { + kerberos_service_name = boost::optional(cct->_conf->rgw_kafka_sasl_kerberos_service_name); + } + if (topic_kerberos_principal.has_value()) { + kerberos_principal = topic_kerberos_principal; + } + if (topic_kerberos_keytab.has_value()) { + kerberos_keytab = topic_kerberos_keytab; + } + } + if (brokers.has_value()) { broker_list.append(","); @@ -715,7 +822,8 @@ public: } connection_id_t tmp_id(broker_list, user, password, ca_location, mechanism, - use_ssl, verify_ssl, ssl_certificate, ssl_key, ssl_key_password); + use_ssl, verify_ssl, ssl_certificate, ssl_key, ssl_key_password, + kerberos_service_name, kerberos_principal, kerberos_keytab); std::lock_guard lock(connections_lock); const auto it = connections.find(tmp_id); // note that ssl vs. non-ssl connection to the same host are two separate connections @@ -734,7 +842,9 @@ public: return false; } - auto conn = std::make_unique(cct, broker_list, use_ssl, verify_ssl, ca_location, user, password, mechanism, ssl_certificate, ssl_key, ssl_key_password); + auto conn = std::make_unique(cct, broker_list, use_ssl, verify_ssl, ca_location, user, password, + mechanism, ssl_certificate, ssl_key, ssl_key_password, + kerberos_service_name, kerberos_principal, kerberos_keytab); if (!new_producer(conn.get())) { ldout(cct, 10) << "Kafka connect: producer creation failed in new connection" << dendl; return false; @@ -856,12 +966,16 @@ bool connect(connection_id_t& conn_id, boost::optional brokers, boost::optional ssl_certificate, boost::optional ssl_key, - boost::optional ssl_key_password) { + boost::optional ssl_key_password, + boost::optional kerberos_service_name, + boost::optional kerberos_principal, + boost::optional kerberos_keytab) { std::shared_lock lock(s_manager_mutex); if (!s_manager) return false; return s_manager->connect(conn_id, url, use_ssl, verify_ssl, ca_location, mechanism, user_name, password, brokers, - ssl_certificate, ssl_key, ssl_key_password); + ssl_certificate, ssl_key, ssl_key_password, + kerberos_service_name, kerberos_principal, kerberos_keytab); } int publish(const connection_id_t& conn_id, diff --git a/src/rgw/rgw_kafka.h b/src/rgw/rgw_kafka.h index 34d06b0d223..52b70b3f420 100644 --- a/src/rgw/rgw_kafka.h +++ b/src/rgw/rgw_kafka.h @@ -31,6 +31,9 @@ struct connection_id_t { std::string ssl_certificate; std::string ssl_key; std::string ssl_key_password; + std::string kerberos_service_name; + std::string kerberos_principal; + std::string kerberos_keytab; bool ssl = false; bool verify_ssl = true; connection_id_t() = default; @@ -43,7 +46,10 @@ struct connection_id_t { bool _verify_ssl, const boost::optional& _ssl_certificate, const boost::optional& _ssl_key, - const boost::optional& _ssl_key_password); + const boost::optional& _ssl_key_password, + const boost::optional& _kerberos_service_name, + const boost::optional& _kerberos_principal, + const boost::optional& _kerberos_keytab); }; std::string to_string(const connection_id_t& id); @@ -60,7 +66,10 @@ bool connect(connection_id_t& conn_id, boost::optional brokers, boost::optional ssl_certificate, boost::optional ssl_key, - boost::optional ssl_key_password); + boost::optional ssl_key_password, + boost::optional kerberos_service_name, + boost::optional kerberos_principal, + boost::optional kerberos_keytab); // publish a message over a connection that was already created int publish(const connection_id_t& conn_id, diff --git a/src/rgw/rgw_rest_pubsub.cc b/src/rgw/rgw_rest_pubsub.cc index 23d25e19753..72b4dac73e8 100644 --- a/src/rgw/rgw_rest_pubsub.cc +++ b/src/rgw/rgw_rest_pubsub.cc @@ -825,7 +825,8 @@ class RGWPSSetTopicAttributesOp : public RGWOp { "verify-ssl", "use-ssl", "ca-location", "amqp-ack-level", "amqp-exchange", "kafka-ack-level", "mechanism", "cloudevents", "user-name", "password", - "ssl-certificate-location", "ssl-key-location", "ssl-key-password"}; + "ssl-certificate-location", "ssl-key-location", "ssl-key-password", + "sasl-kerberos-service-name", "sasl-kerberos-principal", "sasl-kerberos-keytab"}; if (std::find(args.begin(), args.end(), attribute_name) != args.end()) { replace_str(attribute_name, s->info.args.get("AttributeValue")); return 0; -- 2.47.3