Merge pull request ceph#48181 from huynp1999/wip-rgw-kafka-add-sasl-mechanism

src/rgw: SASL mechanism implem

Reviewed-By: ivancich, anthonyeleven, yuvalif

Author: yuvalif

doc/radosgw/notifications.rst

@@ -156,6 +156,7 @@ updating, use the name of an existing topic and different endpoint values).
    [&Attributes.entry.8.key=push-endpoint&Attributes.entry.8.value=<endpoint>]
    [&Attributes.entry.9.key=persistent&Attributes.entry.9.value=true|false]
    [&Attributes.entry.10.key=cloudevents&Attributes.entry.10.value=true|false]
+   [&Attributes.entry.11.key=mechanism&Attributes.entry.11.value=<mechanism>]
 
 Request parameters:
 
@@ -211,10 +212,19 @@ Request parameters:
  - ``ca-location``: If this is provided and a secure connection is used, the
    specified CA will be used instead of the default CA to authenticate the
    broker. 
- - user/password: This must be provided only over HTTPS. Topic creation
-   requests will otherwise be rejected.
- - user/password: This must be provided along with ``use-ssl``. Connections to
-   the broker will otherwise fail.
+ - user/password may be provided over HTTPS. If not, the config parameter
+   `rgw_allow_notification_secrets_in_cleartext` must be `true` in order to create topic
+ - user/password may be provided along with ``use-ssl``.
+   The broker credentials will otherwise be sent over insecure transport
+ - ``mechanism`` may be provided together with user/password (default: ``PLAIN``).
+   The supported SASL mechanisms are:
+
+  - PLAIN
+  - SCRAM-SHA-256
+  - SCRAM-SHA-512
+  - GSSAPI
+  - OAUTHBEARER
+
  - port: This defaults to 9092.
  - kafka-ack-level: No end2end acking is required. Messages may persist in the
    broker before being delivered to their final destinations. Two ack methods

src/rgw/driver/rados/rgw_pubsub_push.cc

@@ -309,7 +309,7 @@ class RGWPubSubKafkaEndpoint : public RGWPubSubEndpoint {
       CephContext* _cct) : 
         cct(_cct),
         topic(_topic),
-        conn(kafka::connect(_endpoint, get_bool(args, "use-ssl", false), get_bool(args, "verify-ssl", true), args.get_optional("ca-location"))) ,
+        conn(kafka::connect(_endpoint, get_bool(args, "use-ssl", false), get_bool(args, "verify-ssl", true), args.get_optional("ca-location"), args.get_optional("mechanism"))) ,
         ack_level(get_ack_level(args)) {
     if (!conn) { 
       throw configuration_error("Kafka: failed to create connection to: " + _endpoint);

src/rgw/rgw_kafka.cc

@@ -73,6 +73,7 @@ struct connection_t {
   const boost::optional<std::string> ca_location;
   const std::string user;
   const std::string password;
+  const boost::optional<std::string> mechanism;
   utime_t timestamp = ceph_clock_now();
 
   // cleanup of all internal connection resource
@@ -107,8 +108,8 @@ struct connection_t {
   // ctor for setting immutable values
   connection_t(CephContext* _cct, const std::string& _broker, bool _use_ssl, bool _verify_ssl, 
           const boost::optional<const std::string&>& _ca_location,
-          const std::string& _user, const std::string& _password) :
-      cct(_cct), broker(_broker), use_ssl(_use_ssl), verify_ssl(_verify_ssl), ca_location(_ca_location), user(_user), password(_password) {}
+          const std::string& _user, const std::string& _password, const boost::optional<const std::string&>& _mechanism) :
+      cct(_cct), broker(_broker), use_ssl(_use_ssl), verify_ssl(_verify_ssl), ca_location(_ca_location), user(_user), password(_password), mechanism(_mechanism) {}                                                                                                                                                        
 
   // dtor also destroys the internals
   ~connection_t() {
@@ -124,6 +125,7 @@ std::string to_string(const connection_ptr_t& conn) {
     str += "\nBroker: " + conn->broker; 
     str += conn->use_ssl ? "\nUse SSL" : ""; 
     str += conn->ca_location ? "\nCA Location: " + *(conn->ca_location) : "";
+    str += conn->mechanism ? "\nSASL Mechanism: " + *(conn->mechanism) : "";
     return str;
 }
 // these are required interfaces so that connection_t could be used inside boost::intrusive_ptr
@@ -220,10 +222,18 @@ connection_ptr_t& create_connection(connection_ptr_t& conn) {
     if (!conn->user.empty()) {
       // use SSL+SASL
       if (rd_kafka_conf_set(conn->temp_conf, "security.protocol", "SASL_SSL", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
-              rd_kafka_conf_set(conn->temp_conf, "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
               rd_kafka_conf_set(conn->temp_conf, "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
               rd_kafka_conf_set(conn->temp_conf, "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) {
+        if (rd_kafka_conf_set(conn->temp_conf, "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(conn->temp_conf, "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+        ldout(conn->cct, 20) << "Kafka connect: using default SASL mechanism" << dendl;
+      }
+
     } else {
       // use only SSL
       if (rd_kafka_conf_set(conn->temp_conf, "security.protocol", "SSL", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
@@ -239,6 +249,20 @@ connection_ptr_t& create_connection(connection_ptr_t& conn) {
     // if (rd_kafka_conf_set(conn->temp_conf, "enable.ssl.certificate.verification", "0", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
 
     ldout(conn->cct, 20) << "Kafka connect: successfully configured security" << dendl;
+  } else if (!conn->user.empty()) {
+      // use SASL+PLAINTEXT
+      if (rd_kafka_conf_set(conn->temp_conf, "security.protocol", "SASL_PLAINTEXT", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
+              rd_kafka_conf_set(conn->temp_conf, "sasl.username", conn->user.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK ||
+              rd_kafka_conf_set(conn->temp_conf, "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(conn->temp_conf, "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(conn->temp_conf, "sasl.mechanism", "PLAIN", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error;
+        ldout(conn->cct, 20) << "Kafka connect: using default SASL mechanism" << dendl;
+      }
   }
 
   // set the global callback for delivery success/fail
@@ -286,9 +310,10 @@ connection_ptr_t create_new_connection(const std::string& broker, CephContext* c
         bool verify_ssl,
         boost::optional<const std::string&> ca_location, 
         const std::string& user, 
-        const std::string& password) { 
+        const std::string& password,
+        boost::optional<const std::string&> mechanism) { 
   // create connection state
-  connection_ptr_t conn(new connection_t(cct, broker, use_ssl, verify_ssl, ca_location, user, password));
+  connection_ptr_t conn(new connection_t(cct, broker, use_ssl, verify_ssl, ca_location, user, password, mechanism));
   return create_connection(conn);
 }
 
@@ -529,7 +554,8 @@ class Manager {
   connection_ptr_t connect(const std::string& url, 
           bool use_ssl,
           bool verify_ssl,
-          boost::optional<const std::string&> ca_location) {
+          boost::optional<const std::string&> ca_location,
+          boost::optional<const std::string&> mechanism) {
     if (stopped) {
       // TODO: increment counter
       ldout(cct, 1) << "Kafka connect: manager is stopped" << dendl;
@@ -548,7 +574,7 @@ class Manager {
     // this should be validated by the regex in parse_url()
     ceph_assert(user.empty() == password.empty());
 
-	if (!user.empty() && !use_ssl) {
+	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 nullptr;
 	}
@@ -568,7 +594,7 @@ class Manager {
       ldout(cct, 1) << "Kafka connect: max connections exceeded" << dendl;
       return nullptr;
     }
-    const auto conn = create_new_connection(broker, cct, use_ssl, verify_ssl, ca_location, user, password);
+    const auto conn = create_new_connection(broker, cct, use_ssl, verify_ssl, ca_location, user, password, mechanism);
     // create_new_connection must always return a connection object
     // even if error occurred during creation. 
     // in such a case the creation will be retried in the main thread
@@ -671,9 +697,10 @@ void shutdown() {
 }
 
 connection_ptr_t connect(const std::string& url, bool use_ssl, bool verify_ssl,
-        boost::optional<const std::string&> ca_location) {
+        boost::optional<const std::string&> ca_location,
+        boost::optional<const std::string&> mechanism) {
   if (!s_manager) return nullptr;
-  return s_manager->connect(url, use_ssl, verify_ssl, ca_location);
+  return s_manager->connect(url, use_ssl, verify_ssl, ca_location, mechanism);
 }
 
 int publish(connection_ptr_t& conn, 

src/rgw/rgw_kafka.h

@@ -31,7 +31,7 @@ bool init(CephContext* cct);
 void shutdown();
 
 // connect to a kafka endpoint
-connection_ptr_t connect(const std::string& url, bool use_ssl, bool verify_ssl, boost::optional<const std::string&> ca_location);
+connection_ptr_t connect(const std::string& url, bool use_ssl, bool verify_ssl, boost::optional<const std::string&> ca_location, boost::optional<const std::string&> mechanism);
 
 // publish a message over a connection that was already created
 int publish(connection_ptr_t& conn,