rgw/kafka: add support for GSSAPI mechanism for Kafka Bucket Notification Endpoint

Signed-off-by: sujay-d07 <sujaydongre07@gmail.com>
This commit is contained in:
sujay-d07 2026-05-19 01:06:09 +05:30
parent 7d65abc9c1
commit 1e9997d9be
6 changed files with 199 additions and 38 deletions

View File

@ -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=<file path>]
[&Attributes.entry.20.key=ssl-key-location&Attributes.entry.20.value=<file path>]
[&Attributes.entry.21.key=ssl-key-password&Attributes.entry.21.value=<password-string>]
[&Attributes.entry.22.key=sasl-kerberos-service-name&Attributes.entry.22.value=<kerberos-service-name>]
[&Attributes.entry.23.key=sasl-kerberos-principal&Attributes.entry.23.value=<kerberos-principal>]
[&Attributes.entry.24.key=sasl-kerberos-keytab&Attributes.entry.24.value=<kerberos-keytab-path>]
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
~~~~~~~~~~~~~

View File

@ -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

View File

@ -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);
}

View File

@ -80,7 +80,10 @@ connection_id_t::connection_id_t(
bool _verify_ssl,
const boost::optional<const std::string&>& _ssl_certificate,
const boost::optional<const std::string&>& _ssl_key,
const boost::optional<const std::string&>& _ssl_key_password
const boost::optional<const std::string&>& _ssl_key_password,
const boost::optional<const std::string&>& _kerberos_service_name,
const boost::optional<const std::string&>& _kerberos_principal,
const boost::optional<const std::string&>& _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<std::string> ssl_certificate;
const boost::optional<std::string> ssl_key;
const boost::optional<std::string> ssl_key_password;
const boost::optional<std::string> kerberos_service_name;
const boost::optional<std::string> kerberos_principal;
const boost::optional<std::string> 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<const std::string&>& _mechanism,
const boost::optional<const std::string&>& _ssl_certificate,
const boost::optional<const std::string&>& _ssl_key,
const boost::optional<const std::string&>& _ssl_key_password) :
const boost::optional<const std::string&>& _ssl_key_password,
const boost::optional<const std::string&>& _kerberos_service_name,
const boost::optional<const std::string&>& _kerberos_principal,
const boost::optional<const std::string&>& _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<const std::string&> brokers,
boost::optional<const std::string&> ssl_certificate,
boost::optional<const std::string&> ssl_key,
boost::optional<const std::string&> ssl_key_password) {
boost::optional<const std::string&> ssl_key_password,
boost::optional<const std::string&> topic_kerberos_service_name,
boost::optional<const std::string&> topic_kerberos_principal,
boost::optional<const std::string&> 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<bool>("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<bool>("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<const std::string&> kerberos_service_name;
boost::optional<const std::string&> kerberos_principal;
boost::optional<const std::string&> 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<const std::string&>(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<connection_t>(cct, broker_list, use_ssl, verify_ssl, ca_location, user, password, mechanism, ssl_certificate, ssl_key, ssl_key_password);
auto conn = std::make_unique<connection_t>(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<const std::string&> brokers,
boost::optional<const std::string&> ssl_certificate,
boost::optional<const std::string&> ssl_key,
boost::optional<const std::string&> ssl_key_password) {
boost::optional<const std::string&> ssl_key_password,
boost::optional<const std::string&> kerberos_service_name,
boost::optional<const std::string&> kerberos_principal,
boost::optional<const std::string&> 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,

View File

@ -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<const std::string&>& _ssl_certificate,
const boost::optional<const std::string&>& _ssl_key,
const boost::optional<const std::string&>& _ssl_key_password);
const boost::optional<const std::string&>& _ssl_key_password,
const boost::optional<const std::string&>& _kerberos_service_name,
const boost::optional<const std::string&>& _kerberos_principal,
const boost::optional<const std::string&>& _kerberos_keytab);
};
std::string to_string(const connection_id_t& id);
@ -60,7 +66,10 @@ bool connect(connection_id_t& conn_id,
boost::optional<const std::string&> brokers,
boost::optional<const std::string&> ssl_certificate,
boost::optional<const std::string&> ssl_key,
boost::optional<const std::string&> ssl_key_password);
boost::optional<const std::string&> ssl_key_password,
boost::optional<const std::string&> kerberos_service_name,
boost::optional<const std::string&> kerberos_principal,
boost::optional<const std::string&> kerberos_keytab);
// publish a message over a connection that was already created
int publish(const connection_id_t& conn_id,

View File

@ -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;