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

Signed-off-by: sujay-d07 <sujaydongre07@gmail.com>

diff --git a/doc/radosgw/notifications.rst b/doc/radosgw/notifications.rst
index de07940ef619fd1c0f93380b6fdcc111db58b66e..2587501a5f0caac810ae781f1c6f690310f324e9 100644 (file)
--- 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=<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
 ~~~~~~~~~~~~~

diff --git a/src/common/options/rgw.yaml.in b/src/common/options/rgw.yaml.in
index 3888e2f92b9514fd65b7215b4784ae3f4e7f15f6..921069cfdd0ea61bfd31acb9b4cdc9d2d3f2144e 100644 (file)
--- 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 3fdf2e9c5023cd1921215e70fdd9432c7e14d2d9..69b9d3ae240749c8e273ba3ae0a380fcfa7b7c2e 100644 (file)
--- 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 c86b8445723a9e0e54e746107102b26a8a413ef9..6043c92b4bdc87397b1e0d3da288aff7e0c34b81 100644 (file)
--- 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<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,

diff --git a/src/rgw/rgw_kafka.h b/src/rgw/rgw_kafka.h
index 34d06b0d223102be841fa855550042c5f9c14c4e..52b70b3f4209fd99042376ea42b5cfa2cc09a289 100644 (file)
--- 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<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,

diff --git a/src/rgw/rgw_rest_pubsub.cc b/src/rgw/rgw_rest_pubsub.cc
index 23d25e197536f7c2189de84e129bff0d3296209b..72b4dac73e8bf18b61849ec6d1cf82bf3918195b 100644 (file)
--- 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;