src/rgw: SASL_PLAINTEXT implementation

SASL_PLAINTEXT will be supported by the configuration parameter rgw_allow_notification_secrets_in_cleartext
Signed-off-by: Huy Nguyen <huynp1999@gmail.com>

diff --git a/doc/radosgw/notifications.rst b/doc/radosgw/notifications.rst
index f7244f63503c46ddfea1547ac87aa6fdfff67154..89ebd357d7e8d93b7e92aea595b48b04d689c13e 100644 (file)
--- a/doc/radosgw/notifications.rst
+++ b/doc/radosgw/notifications.rst
@@ -212,11 +212,12 @@ 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.
- - mechanism: may be provided together with user/password (default: ``PLAIN``). The supported SASL mechanisms are:
+ - 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

diff --git a/src/rgw/rgw_kafka.cc b/src/rgw/rgw_kafka.cc
index 26934c303a198e87f4766728416cc4906936b2b9..ef647a601a3e30c56f84436b6ce005b0b1d4c351 100644 (file)
--- a/src/rgw/rgw_kafka.cc
+++ b/src/rgw/rgw_kafka.cc
@@ -249,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
@@ -560,7 +574,7 @@ public:
     // 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;
        }