Merge pull request ceph#56493 from igomon-bloomberg/wip_kafka_sasl_username_password_attrs

rgw/s3-notifications: use user-name/password topic attributes for SASL authentication

Reviewed-by: Yuval Lifshitz <[email protected]>

Author: ivancich

doc/radosgw/notifications.rst

@@ -164,6 +164,8 @@ updating, use the name of an existing topic and different endpoint values).
    [&Attributes.entry.13.key=max_retries&Attributes.entry.13.value=<retries number>]
    [&Attributes.entry.14.key=retry_sleep_duration&Attributes.entry.14.value=<sleep seconds>]
    [&Attributes.entry.15.key=Policy&Attributes.entry.15.value=<policy-JSON-string>]
+   [&Attributes.entry.16.key=user-name&Attributes.entry.16.value=<user-name-string>]
+   [&Attributes.entry.17.key=password&Attributes.entry.17.value=<password-string>]
 
 Request parameters:
 
@@ -252,6 +254,10 @@ Request parameters:
  - user/password: This should be provided over HTTPS. If not, the config parameter `rgw_allow_notification_secrets_in_cleartext` must be `true` in order to create topics.
  - user/password: This should be provided together with ``use-ssl``. If not, the broker credentials will be sent over insecure transport.
  - mechanism: may be provided together with user/password (default: ``PLAIN``). The supported SASL mechanisms are:
+ - ``user-name``: User name to use when connecting to the Kafka broker. If both this parameter and URI user are provided then this parameter overrides the URI user.
+    The same security considerations are in place for this parameter as are for user/password.
+ - ``password``: Password to use when connecting to the Kafka broker. If both this parameter and URI password are provided then this parameter overrides the URI password.
+    The same security considerations are in place for this parameter as are for user/password.
 
   - PLAIN
   - SCRAM-SHA-256

src/rgw/driver/rados/rgw_pubsub_push.cc

@@ -303,8 +303,13 @@ class RGWPubSubKafkaEndpoint : public RGWPubSubEndpoint {
         cct(_cct),
         topic(_topic),
         ack_level(get_ack_level(args)) {
-    if (!kafka::connect(conn_name, _endpoint, get_bool(args, "use-ssl", false), get_bool(args, "verify-ssl", true), 
-          args.get_optional("ca-location"), args.get_optional("mechanism"))) {
+    if (!kafka::connect(conn_name, _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"))) {
       throw configuration_error("Kafka: failed to create connection to: " + _endpoint);
     }
   }

src/rgw/rgw_kafka.cc

@@ -554,7 +554,9 @@ class Manager {
           bool use_ssl,
           bool verify_ssl,
           boost::optional<const std::string&> ca_location,
-          boost::optional<const std::string&> mechanism) {
+          boost::optional<const std::string&> mechanism,
+          boost::optional<const std::string&> topic_user_name,
+          boost::optional<const std::string&> topic_password) {
     if (stopped) {
       ldout(cct, 1) << "Kafka connect: manager is stopped" << dendl;
       return false;
@@ -568,6 +570,21 @@ class Manager {
       return false;
     }
 
+    // check if username/password was already supplied via topic attributes
+    // and if also provided as part of the endpoint URL issue a warning
+    if (topic_user_name.has_value()) {
+      if (!user.empty()) {
+        ldout(cct, 5) << "Kafka connect: username provided via both topic attributes and endpoint URL: using topic attributes" << dendl;
+      }
+      user = topic_user_name.get();
+    }
+    if (topic_password.has_value()) {
+      if (!password.empty()) {
+        ldout(cct, 5) << "Kafka connect: password provided via both topic attributes and endpoint URL: using topic attributes" << dendl;
+      }
+      password = topic_password.get();
+    }
+
     // this should be validated by the regex in parse_url()
     ceph_assert(user.empty() == password.empty());
 
@@ -694,9 +711,11 @@ void shutdown() {
 
 bool connect(std::string& broker, const std::string& url, bool use_ssl, bool verify_ssl,
         boost::optional<const std::string&> ca_location,
-        boost::optional<const std::string&> mechanism) {
+        boost::optional<const std::string&> mechanism,
+        boost::optional<const std::string&> user_name,
+        boost::optional<const std::string&> password) {
   if (!s_manager) return false;
-  return s_manager->connect(broker, url, use_ssl, verify_ssl, ca_location, mechanism);
+  return s_manager->connect(broker, url, use_ssl, verify_ssl, ca_location, mechanism, user_name, password);
 }
 
 int publish(const std::string& conn_name,

src/rgw/rgw_kafka.h

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

src/rgw/rgw_rest_pubsub.cc

@@ -37,7 +37,7 @@ bool verify_transport_security(CephContext *cct, const RGWEnv& env) {
 // make sure that if user/password are passed inside URL, it is over secure connection
 // update rgw_pubsub_dest to indicate that a password is stored in the URL
 bool validate_and_update_endpoint_secret(rgw_pubsub_dest& dest, CephContext *cct,
-                                         const RGWEnv& env, std::string& message)
+                                         const req_info& ri, std::string& message)
 {
   if (dest.push_endpoint.empty()) {
     return true;
@@ -48,11 +48,31 @@ bool validate_and_update_endpoint_secret(rgw_pubsub_dest& dest, CephContext *cct
     message = "Malformed URL for push-endpoint";
     return false;
   }
+
+  const auto& args=ri.args;
+  auto topic_user_name=args.get_optional("user-name");
+  auto topic_password=args.get_optional("password");
+
+  // check if username/password was already supplied via topic attributes
+  // and if also provided as part of the endpoint URL issue a warning
+  if (topic_user_name.has_value()) {
+    if (!user.empty()) {
+      message = "Username provided via both topic attributes and endpoint URL: using topic attributes";
+    }
+    user = topic_user_name.get();
+  }
+  if (topic_password.has_value()) {
+    if (!password.empty()) {
+      message = "Password provided via both topic attributes and endpoint URL: using topic attributes";
+    }
+    password = topic_password.get();
+  }
+
   // this should be verified inside parse_url()
   ceph_assert(user.empty() == password.empty());
   if (!user.empty()) {
     dest.stored_secret = true;
-    if (!verify_transport_security(cct, env)) {
+    if (!verify_transport_security(cct, *ri.env)) {
       message = "Topic contains secrets that must be transmitted over a secure transport";
       return false;
     }
@@ -241,7 +261,7 @@ class RGWPSCreateTopicOp : public RGWOp {
     s->info.args.get_int("max_retries", reinterpret_cast<int *>(&dest.max_retries), rgw::notify::DEFAULT_GLOBAL_VALUE);
     s->info.args.get_int("retry_sleep_duration", reinterpret_cast<int *>(&dest.retry_sleep_duration), rgw::notify::DEFAULT_GLOBAL_VALUE);
 
-    if (!validate_and_update_endpoint_secret(dest, s->cct, *s->info.env, s->err.message)) {
+    if (!validate_and_update_endpoint_secret(dest, s->cct, s->info, s->err.message)) {
       return -EINVAL;
     }
     // Store topic Policy.
@@ -729,7 +749,7 @@ class RGWPSSetTopicAttributesOp : public RGWOp {
                            rgw::notify::DEFAULT_GLOBAL_VALUE);
     } else if (attribute_name == "push-endpoint") {
       dest.push_endpoint = s->info.args.get("AttributeValue");
-      if (!validate_and_update_endpoint_secret(dest, s->cct, *s->info.env, s->err.message)) {
+      if (!validate_and_update_endpoint_secret(dest, s->cct, s->info, s->err.message)) {
         return -EINVAL;
       }
     } else if (attribute_name == "Policy") {
@@ -755,7 +775,8 @@ class RGWPSSetTopicAttributesOp : public RGWOp {
       };
       static constexpr std::initializer_list<const char*> args = {
           "verify-ssl",    "use-ssl",         "ca-location", "amqp-ack-level",
-          "amqp-exchange", "kafka-ack-level", "mechanism",   "cloudevents"};
+          "amqp-exchange", "kafka-ack-level", "mechanism",   "cloudevents",
+          "user-name",     "password"};
       if (std::find(args.begin(), args.end(), attribute_name) != args.end()) {
         replace_str(attribute_name, s->info.args.get("AttributeValue"));
         return 0;

src/rgw/rgw_rest_s3.cc

@@ -5174,8 +5174,6 @@ void update_attribute_map(const std::string& input, AttributeMap& map) {
 void parse_post_action(const std::string& post_body, req_state* s)
 {
   if (post_body.size() > 0) {
-    ldpp_dout(s, 10) << "Content of POST: " << post_body << dendl;
-
     if (post_body.find("Action") != string::npos) {
       const boost::char_separator<char> sep("&");
       const boost::tokenizer<boost::char_separator<char>> tokens(post_body, sep);

src/test/rgw/bucket_notification/test_bn.py

@@ -4389,7 +4389,7 @@ def test_ps_s3_topic_no_permissions():
     conn2.delete_bucket(bucket_name)
 
 
-def kafka_security(security_type, mechanism='PLAIN'):
+def kafka_security(security_type, mechanism='PLAIN', use_topic_attrs_for_creds=False):
     """ test pushing kafka s3 notification securly to master """
     conn = connection()
     zonegroup = get_config_zonegroup()
@@ -4400,7 +4400,10 @@ def kafka_security(security_type, mechanism='PLAIN'):
     topic_name = bucket_name+'_topic'
     # create s3 topic
     if security_type == 'SASL_SSL':
-        endpoint_address = 'kafka://alice:alice-secret@' + kafka_server + ':9094'
+        if not use_topic_attrs_for_creds:
+            endpoint_address = 'kafka://alice:alice-secret@' + kafka_server + ':9094'
+        else:
+            endpoint_address = 'kafka://' + kafka_server + ':9094'
     elif security_type == 'SSL':
         endpoint_address = 'kafka://' + kafka_server + ':9093'
     elif security_type == 'SASL_PLAINTEXT':
@@ -4413,6 +4416,8 @@ def kafka_security(security_type, mechanism='PLAIN'):
     elif security_type == 'SASL_SSL':
         KAFKA_DIR = os.environ['KAFKA_DIR']
         endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=true&ca-location='+KAFKA_DIR+'/y-ca.crt&mechanism='+mechanism
+        if use_topic_attrs_for_creds:
+            endpoint_args += '&user-name=alice&password=alice-secret'
     else:
         KAFKA_DIR = os.environ['KAFKA_DIR']
         endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=true&ca-location='+KAFKA_DIR+'/y-ca.crt'
@@ -4485,6 +4490,11 @@ def test_ps_s3_notification_push_kafka_security_ssl_sasl():
     kafka_security('SASL_SSL')
 
 
+@attr('kafka_security_test')
+def test_ps_s3_notification_push_kafka_security_ssl_sasl_attrs():
+    kafka_security('SASL_SSL', use_topic_attrs_for_creds=True)
+
+
 @attr('kafka_security_test')
 def test_ps_s3_notification_push_kafka_security_sasl():
     kafka_security('SASL_PLAINTEXT')