diff options
Diffstat (limited to 'src/rgw')
| -rw-r--r-- | src/rgw/CMakeLists.txt | 3 | ||||
| -rw-r--r-- | src/rgw/rgw_kafka.cc | 166 | ||||
| -rw-r--r-- | src/rgw/rgw_kafka.h | 6 | ||||
| -rw-r--r-- | src/rgw/rgw_pubsub.h | 9 | ||||
| -rw-r--r-- | src/rgw/rgw_pubsub_push.cc | 75 | ||||
| -rw-r--r-- | src/rgw/rgw_rest_pubsub.cc | 61 | ||||
| -rw-r--r-- | src/rgw/rgw_rest_pubsub_common.cc | 57 | ||||
| -rw-r--r-- | src/rgw/rgw_rest_pubsub_common.h | 5 | ||||
| -rw-r--r-- | src/rgw/rgw_sync_module_pubsub_rest.cc | 9 | ||||
| -rw-r--r-- | src/rgw/rgw_url.cc | 48 | ||||
| -rw-r--r-- | src/rgw/rgw_url.h | 12 |
11 files changed, 335 insertions, 116 deletions
diff --git a/src/rgw/CMakeLists.txt b/src/rgw/CMakeLists.txt index d36d57adad6..1c95aed660d 100644 --- a/src/rgw/CMakeLists.txt +++ b/src/rgw/CMakeLists.txt @@ -142,7 +142,8 @@ set(librgw_common_srcs rgw_perf_counters.cc rgw_rest_iam.cc rgw_object_lock.cc - rgw_kms.cc) + rgw_kms.cc + rgw_url.cc) if(WITH_RADOSGW_AMQP_ENDPOINT) list(APPEND librgw_common_srcs rgw_amqp.cc) diff --git a/src/rgw/rgw_kafka.cc b/src/rgw/rgw_kafka.cc index 4f7751ae6c6..dfaefdfb270 100644 --- a/src/rgw/rgw_kafka.cc +++ b/src/rgw/rgw_kafka.cc @@ -1,13 +1,12 @@ // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- // vim: ts=8 sw=2 smarttab ft=cpp -//#include "include/compat.h" #include "rgw_kafka.h" +#include "rgw_url.h" #include <librdkafka/rdkafka.h> #include "include/ceph_assert.h" #include <sstream> #include <cstring> -#include <regex> #include <unordered_map> #include <string> #include <vector> @@ -32,17 +31,14 @@ bool operator==(const rd_kafka_topic_t* rkt, const std::string& name) { namespace rgw::kafka { // status codes for publishing +// TODO: use the actual error code (when conn exists) instead of STATUS_CONNECTION_CLOSED when replying to client static const int STATUS_CONNECTION_CLOSED = -0x1002; static const int STATUS_QUEUE_FULL = -0x1003; static const int STATUS_MAX_INFLIGHT = -0x1004; static const int STATUS_MANAGER_STOPPED = -0x1005; // status code for connection opening -static const int STATUS_CONF_ALLOC_FAILED = -0x2001; -static const int STATUS_GET_BROKER_LIST_FAILED = -0x2002; -static const int STATUS_CREATE_PRODUCER_FAILED = -0x2003; +static const int STATUS_CONF_ALLOC_FAILED = -0x2001; -static const int STATUS_CREATE_TOPIC_FAILED = -0x3008; -static const int NO_REPLY_CODE = 0x0; static const int STATUS_OK = 0x0; // struct for holding the callback and its tag in the callback list @@ -70,8 +66,14 @@ struct connection_t { uint64_t delivery_tag = 1; int status; mutable std::atomic<int> ref_count = 0; - CephContext* cct = nullptr; + CephContext* const cct; CallbackList callbacks; + const std::string broker; + const bool use_ssl; + const bool verify_ssl; // TODO currently iognored, not supported in librdkafka v0.11.6 + const boost::optional<std::string> ca_location; + const std::string user; + const std::string password; // cleanup of all internal connection resource // the object can still remain, and internal connection @@ -102,6 +104,12 @@ struct connection_t { return (producer != nullptr && !marked_for_deletion); } + // 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) {} + // dtor also destroys the internals ~connection_t() { destroy(STATUS_CONNECTION_CLOSED); @@ -111,6 +119,13 @@ struct connection_t { friend void intrusive_ptr_release(const connection_t* p); }; +std::string to_string(const connection_ptr_t& conn) { + std::string str; + str += "\nBroker: " + conn->broker; + str += conn->use_ssl ? "\nUse SSL" : ""; + str += conn->ca_location ? "\nCA Location: " + *(conn->ca_location) : ""; + return str; +} // these are required interfaces so that connection_t could be used inside boost::intrusive_ptr void intrusive_ptr_add_ref(const connection_t* p) { ++p->ref_count; @@ -124,6 +139,8 @@ void intrusive_ptr_release(const connection_t* p) { // convert int status to string - including RGW specific values std::string status_to_string(int s) { switch (s) { + case STATUS_OK: + return "STATUS_OK"; case STATUS_CONNECTION_CLOSED: return "RGW_KAFKA_STATUS_CONNECTION_CLOSED"; case STATUS_QUEUE_FULL: @@ -134,13 +151,8 @@ std::string status_to_string(int s) { return "RGW_KAFKA_STATUS_MANAGER_STOPPED"; case STATUS_CONF_ALLOC_FAILED: return "RGW_KAFKA_STATUS_CONF_ALLOC_FAILED"; - case STATUS_CREATE_PRODUCER_FAILED: - return "STATUS_CREATE_PRODUCER_FAILED"; - case STATUS_CREATE_TOPIC_FAILED: - return "STATUS_CREATE_TOPIC_FAILED"; } - // TODO: how to handle "s" in this case? - return std::string(rd_kafka_err2str(rd_kafka_last_error())); + return std::string(rd_kafka_err2str((rd_kafka_resp_err_t)s)); } void message_callback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* opaque) { @@ -160,7 +172,7 @@ void message_callback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* const auto tag_it = std::find(callbacks_begin, callbacks_end, *tag); if (tag_it != callbacks_end) { ldout(conn->cct, 20) << "Kafka run: n/ack received, invoking callback with tag=" << - *tag << " and result=" << result << dendl; + *tag << " and result=" << rd_kafka_err2str(result) << dendl; tag_it->cb(result); conn->callbacks.erase(tag_it); } else { @@ -173,14 +185,13 @@ void message_callback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage, void* } // utility function to create a connection, when the connection object already exists -connection_ptr_t& create_connection(connection_ptr_t& conn, const std::string& broker) { +connection_ptr_t& create_connection(connection_ptr_t& conn) { // pointer must be valid and not marked for deletion ceph_assert(conn && !conn->marked_for_deletion); // reset all status codes conn->status = STATUS_OK; - - char errstr[512]; + char errstr[512] = {0}; conn->temp_conf = rd_kafka_conf_new(); if (!conn->temp_conf) { @@ -189,38 +200,68 @@ connection_ptr_t& create_connection(connection_ptr_t& conn, const std::string& b } // get list of brokers based on the bootsrap broker - if (rd_kafka_conf_set(conn->temp_conf, "bootstrap.servers", broker.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) { - conn->status = STATUS_GET_BROKER_LIST_FAILED; - // TODO: use errstr - return conn; + if (rd_kafka_conf_set(conn->temp_conf, "bootstrap.servers", conn->broker.c_str(), errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) goto conf_error; + + if (conn->use_ssl) { + 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; + } 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; + ldout(conn->cct, 20) << "Kafka connect: successfully configured SSL security" << dendl; + } + if (conn->ca_location) { + if (rd_kafka_conf_set(conn->temp_conf, "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; + } + // Note: when librdkafka.1.0 is available the following line could be uncommented instead of the callback setting call + // 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; } // set the global callback for delivery success/fail rd_kafka_conf_set_dr_msg_cb(conn->temp_conf, message_callback); // set the global opaque pointer to be the connection itself - rd_kafka_conf_set_opaque (conn->temp_conf, conn.get()); + rd_kafka_conf_set_opaque(conn->temp_conf, conn.get()); // create the producer conn->producer = rd_kafka_new(RD_KAFKA_PRODUCER, conn->temp_conf, errstr, sizeof(errstr)); - if (conn->producer) { - conn->status = STATUS_CREATE_PRODUCER_FAILED; - // TODO: use errstr + if (!conn->producer) { + conn->status = rd_kafka_last_error(); + ldout(conn->cct, 1) << "Kafka connect: failed to create producer: " << errstr << dendl; return conn; } + ldout(conn->cct, 20) << "Kafka connect: successfully created new producer" << dendl; // conf ownership passed to producer conn->temp_conf = nullptr; return conn; -} +conf_error: + conn->status = rd_kafka_last_error(); + ldout(conn->cct, 1) << "Kafka connect: configuration failed: " << errstr << dendl; + return conn; +} // utility function to create a new connection -connection_ptr_t create_new_connection(const std::string& broker, CephContext* cct) { +connection_ptr_t create_new_connection(const std::string& broker, CephContext* cct, + bool use_ssl, + bool verify_ssl, + boost::optional<const std::string&> ca_location, + const std::string& user, + const std::string& password) { // create connection state - connection_ptr_t conn = new connection_t; - conn->cct = cct; - return create_connection(conn, broker); + connection_ptr_t conn(new connection_t(cct, broker, use_ssl, verify_ssl, ca_location, user, password)); + return create_connection(conn); } /// struct used for holding messages in the message queue @@ -236,22 +277,6 @@ struct message_wrapper_t { reply_callback_t _cb) : conn(_conn), topic(_topic), message(_message), cb(_cb) {} }; -// parse a URL of the form: kafka://<host>[:port] -// to a: host[:port] -int parse_url(const std::string& url, std::string& broker) { - std::regex url_regex ( - R"(^(([^:\/?#]+):)?(//([^\/?#]*))?([^?#]*)(\?([^#]*))?(#(.*))?)", - std::regex::extended - ); - const auto HOST_AND_PORT = 4; - std::smatch url_match_result; - if (std::regex_match(url, url_match_result, url_regex)) { - broker = url_match_result[HOST_AND_PORT]; - return 0; - } - return -1; -} - typedef std::unordered_map<std::string, connection_ptr_t> ConnectionList; typedef boost::lockfree::queue<message_wrapper_t*, boost::lockfree::fixed_sized<true>> MessageQueue; @@ -290,9 +315,9 @@ private: if (!conn->is_ok()) { // connection had an issue while message was in the queue // TODO add error stats - ldout(conn->cct, 1) << "Kafka publish: connection had an issue while message was in the queue" << dendl; + ldout(conn->cct, 1) << "Kafka publish: connection had an issue while message was in the queue. error: " << status_to_string(conn->status) << dendl; if (message->cb) { - message->cb(STATUS_CONNECTION_CLOSED); + message->cb(conn->status); } return; } @@ -303,11 +328,12 @@ private: if (topic_it == conn->topics.end()) { topic = rd_kafka_topic_new(conn->producer, message->topic.c_str(), nullptr); if (!topic) { - ldout(conn->cct, 1) << "Kafka publish: failed to create topic: " << message->topic << dendl; + const auto err = rd_kafka_last_error(); + ldout(conn->cct, 1) << "Kafka publish: failed to create topic: " << message->topic << " error: " << status_to_string(err) << dendl; if (message->cb) { - message->cb(STATUS_CREATE_TOPIC_FAILED); + message->cb(err); } - conn->destroy(STATUS_CREATE_TOPIC_FAILED); + conn->destroy(err); return; } // TODO use the topics list as an LRU cache @@ -337,12 +363,13 @@ private: tag); if (rc == -1) { const auto err = rd_kafka_last_error(); - ldout(conn->cct, 1) << "Kafka publish: failed to produce: " << rd_kafka_err2str(err) << dendl; - // TODO: dont error on full queue, retry instead + ldout(conn->cct, 10) << "Kafka publish: failed to produce: " << rd_kafka_err2str(err) << dendl; + // TODO: dont error on full queue, and don't destroy connection, retry instead // immediatly invoke callback on error if needed if (message->cb) { message->cb(err); } + conn->destroy(err); delete tag; } @@ -352,7 +379,7 @@ private: ldout(conn->cct, 20) << "Kafka publish (with callback, tag=" << *tag << "): OK. Queue has: " << q_len << " callbacks" << dendl; conn->callbacks.emplace_back(*tag, message->cb); } else { - // immediately invoke callback with error + // immediately invoke callback with error - this is not a connection error ldout(conn->cct, 1) << "Kafka publish (with callback): failed with error: callback queue full" << dendl; message->cb(STATUS_MAX_INFLIGHT); // tag will be deleted when the global callback is invoked @@ -400,9 +427,10 @@ private: // try to reconnect the connection if it has an error if (!conn->is_ok()) { + ldout(conn->cct, 10) << "Kafka run: connection status is: " << status_to_string(conn->status) << dendl; const auto& broker = conn_it->first; ldout(conn->cct, 20) << "Kafka run: retry connection" << dendl; - if (create_connection(conn, broker)->is_ok() == false) { + if (create_connection(conn)->is_ok() == false) { ldout(conn->cct, 10) << "Kafka run: connection (" << broker << ") retry failed" << dendl; // TODO: add error counter for failed retries // TODO: add exponential backoff for retries @@ -477,7 +505,10 @@ public: } // connect to a broker, or reuse an existing connection if already connected - connection_ptr_t connect(const std::string& url) { + connection_ptr_t connect(const std::string& url, + bool use_ssl, + bool verify_ssl, + boost::optional<const std::string&> ca_location) { if (stopped) { // TODO: increment counter ldout(cct, 1) << "Kafka connect: manager is stopped" << dendl; @@ -485,14 +516,25 @@ public: } std::string broker; - if (0 != parse_url(url, broker)) { + std::string user; + std::string password; + if (!parse_url_authority(url, broker, user, password)) { // TODO: increment counter ldout(cct, 1) << "Kafka connect: URL parsing failed" << dendl; return nullptr; } + // this should be validated by the regex in parse_url() + ceph_assert(user.empty() == password.empty()); + + if (!user.empty() && !use_ssl) { + ldout(cct, 1) << "Kafka connect: user/password are only allowed over secure connection" << dendl; + return nullptr; + } + std::lock_guard lock(connections_lock); const auto it = connections.find(broker); + // note that ssl vs. non-ssl connection to the same host are two separate conenctions if (it != connections.end()) { if (it->second->marked_for_deletion) { // TODO: increment counter @@ -510,14 +552,13 @@ public: ldout(cct, 1) << "Kafka connect: max connections exceeded" << dendl; return nullptr; } - const auto conn = create_new_connection(broker, cct); + const auto conn = create_new_connection(broker, cct, use_ssl, verify_ssl, ca_location, user, password); // 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 ceph_assert(conn); ++connection_count; ldout(cct, 10) << "Kafka connect: new connection is created. Total connections: " << connection_count << dendl; - ldout(cct, 10) << "Kafka connect: new connection status is: " << status_to_string(conn->status) << dendl; return connections.emplace(broker, conn).first->second; } @@ -613,9 +654,10 @@ void shutdown() { s_manager = nullptr; } -connection_ptr_t connect(const std::string& url) { +connection_ptr_t connect(const std::string& url, bool use_ssl, bool verify_ssl, + boost::optional<const std::string&> ca_location) { if (!s_manager) return nullptr; - return s_manager->connect(url); + return s_manager->connect(url, use_ssl, verify_ssl, ca_location); } int publish(connection_ptr_t& conn, diff --git a/src/rgw/rgw_kafka.h b/src/rgw/rgw_kafka.h index 7319679b09c..cccdd65b6ab 100644 --- a/src/rgw/rgw_kafka.h +++ b/src/rgw/rgw_kafka.h @@ -6,6 +6,7 @@ #include <string> #include <functional> #include <boost/smart_ptr/intrusive_ptr.hpp> +#include <boost/optional.hpp> class CephContext; @@ -30,7 +31,7 @@ bool init(CephContext* cct); void shutdown(); // connect to a kafka endpoint -connection_ptr_t connect(const std::string& url); +connection_ptr_t connect(const std::string& url, bool use_ssl, bool verify_ssl, boost::optional<const std::string&> ca_location); // publish a message over a connection that was already created int publish(connection_ptr_t& conn, @@ -73,5 +74,8 @@ size_t get_max_queue(); // disconnect from a kafka broker bool disconnect(connection_ptr_t& conn); +// display connection as string +std::string to_string(const connection_ptr_t& conn); + } diff --git a/src/rgw/rgw_pubsub.h b/src/rgw/rgw_pubsub.h index a6f97ea6ff1..ee975700956 100644 --- a/src/rgw/rgw_pubsub.h +++ b/src/rgw/rgw_pubsub.h @@ -341,19 +341,21 @@ struct rgw_pubsub_sub_dest { std::string push_endpoint; std::string push_endpoint_args; std::string arn_topic; + bool stored_secret = false; void encode(bufferlist& bl) const { - ENCODE_START(3, 1, bl); + ENCODE_START(4, 1, bl); encode(bucket_name, bl); encode(oid_prefix, bl); encode(push_endpoint, bl); encode(push_endpoint_args, bl); encode(arn_topic, bl); + encode(stored_secret, bl); ENCODE_FINISH(bl); } void decode(bufferlist::const_iterator& bl) { - DECODE_START(3, bl); + DECODE_START(4, bl); decode(bucket_name, bl); decode(oid_prefix, bl); decode(push_endpoint, bl); @@ -363,6 +365,9 @@ struct rgw_pubsub_sub_dest { if (struct_v >= 3) { decode(arn_topic, bl); } + if (struct_v >= 4) { + decode(stored_secret, bl); + } DECODE_FINISH(bl); } diff --git a/src/rgw/rgw_pubsub_push.cc b/src/rgw/rgw_pubsub_push.cc index 4d0496687ad..6230330d4a6 100644 --- a/src/rgw/rgw_pubsub_push.cc +++ b/src/rgw/rgw_pubsub_push.cc @@ -220,7 +220,7 @@ private: const std::string topic; amqp::connection_ptr_t conn; const std::string message; - const ack_level_t ack_level; // TODO not used for now + [[maybe_unused]] const ack_level_t ack_level; // TODO not used for now public: AckPublishCR(CephContext* cct, @@ -415,6 +415,7 @@ static const std::string AMQP_1_0("1-0"); static const std::string AMQP_SCHEMA("amqp"); #endif // ifdef WITH_RADOSGW_AMQP_ENDPOINT + #ifdef WITH_RADOSGW_KAFKA_ENDPOINT class RGWPubSubKafkaEndpoint : public RGWPubSubEndpoint { private: @@ -423,11 +424,57 @@ private: Broker, }; CephContext* const cct; - const std::string endpoint; const std::string topic; kafka::connection_ptr_t conn; - ack_level_t ack_level; - std::string str_ack_level; + const ack_level_t ack_level; + + static bool get_verify_ssl(const RGWHTTPArgs& args) { + bool exists; + auto str_verify_ssl = args.get("verify-ssl", &exists); + if (!exists) { + // verify server certificate by default + return true; + } + boost::algorithm::to_lower(str_verify_ssl); + if (str_verify_ssl == "true") { + return true; + } + if (str_verify_ssl == "false") { + return false; + } + throw configuration_error("'verify-ssl' must be true/false, not: " + str_verify_ssl); + } + + static bool get_use_ssl(const RGWHTTPArgs& args) { + bool exists; + auto str_use_ssl = args.get("use-ssl", &exists); + if (!exists) { + // by default ssl not used + return false; + } + boost::algorithm::to_lower(str_use_ssl); + if (str_use_ssl == "true") { + return true; + } + if (str_use_ssl == "false") { + return false; + } + throw configuration_error("'use-ssl' must be true/false, not: " + str_use_ssl); + } + + static ack_level_t get_ack_level(const RGWHTTPArgs& args) { + bool exists; + // get ack level + const auto str_ack_level = args.get("kafka-ack-level", &exists); + if (!exists || str_ack_level == "broker") { + // "broker" is default + return ack_level_t::Broker; + } + if (str_ack_level == "none") { + return ack_level_t::None; + } + throw configuration_error("Kafka: invalid kafka-ack-level: " + str_ack_level); + } // NoAckPublishCR implements async kafka publishing via coroutine // This coroutine ends when it send the message and does not wait for an ack @@ -524,22 +571,11 @@ public: const RGWHTTPArgs& args, CephContext* _cct) : cct(_cct), - endpoint(_endpoint), topic(_topic), - conn(kafka::connect(endpoint)) { + conn(kafka::connect(_endpoint, get_use_ssl(args), get_verify_ssl(args), args.get_optional("ca-location"))) , + ack_level(get_ack_level(args)) { if (!conn) { - throw configuration_error("Kafka: failed to create connection to: " + endpoint); - } - bool exists; - // get ack level - str_ack_level = args.get("kafka-ack-level", &exists); - if (!exists || str_ack_level == "broker") { - // "broker" is default - ack_level = ack_level_t::Broker; - } else if (str_ack_level == "none") { - ack_level = ack_level_t::None; - } else { - throw configuration_error("Kafka: invalid kafka-ack-level: " + str_ack_level); + throw configuration_error("Kafka: failed to create connection to: " + _endpoint); } } @@ -638,9 +674,8 @@ public: std::string to_str() const override { std::string str("Kafka Endpoint"); - str += "\nURI: " + endpoint; + str += kafka::to_string(conn); str += "\nTopic: " + topic; - str += "\nAck Level: " + str_ack_level; return str; } }; diff --git a/src/rgw/rgw_rest_pubsub.cc b/src/rgw/rgw_rest_pubsub.cc index 08d8c544cf1..1f7bce65adf 100644 --- a/src/rgw/rgw_rest_pubsub.cc +++ b/src/rgw/rgw_rest_pubsub.cc @@ -19,6 +19,7 @@ #define dout_context g_ceph_context #define dout_subsys ceph_subsys_rgw + // command (AWS compliant): // POST // Action=CreateTopic&Name=<topic-name>[&push-endpoint=<endpoint>[&<arg1>=<value1>]] @@ -27,21 +28,25 @@ public: int get_params() override { topic_name = s->info.args.get("Name"); if (topic_name.empty()) { - ldout(s->cct, 1) << "CreateTopic Action 'Name' argument is missing" << dendl; - return -EINVAL; + ldout(s->cct, 1) << "CreateTopic Action 'Name' argument is missing" << dendl; + return -EINVAL; } dest.push_endpoint = s->info.args.get("push-endpoint"); + + if (!validate_and_update_endpoint_secret(dest, s->cct, *(s->info.env))) { + return -EINVAL; + } for (const auto param : s->info.args.get_params()) { - if (param.first == "Action" || param.first == "Name" || param.first == "PayloadHash") { - continue; - } - dest.push_endpoint_args.append(param.first+"="+param.second+"&"); + if (param.first == "Action" || param.first == "Name" || param.first == "PayloadHash") { + continue; + } + dest.push_endpoint_args.append(param.first+"="+param.second+"&"); } if (!dest.push_endpoint_args.empty()) { - // remove last separator - dest.push_endpoint_args.pop_back(); + // remove last separator + dest.push_endpoint_args.pop_back(); } // dest object only stores endpoint info @@ -160,8 +165,8 @@ public: const auto topic_arn = rgw::ARN::parse((s->info.args.get("TopicArn"))); if (!topic_arn || topic_arn->resource.empty()) { - ldout(s->cct, 1) << "DeleteTopic Action 'TopicArn' argument is missing or invalid" << dendl; - return -EINVAL; + ldout(s->cct, 1) << "DeleteTopic Action 'TopicArn' argument is missing or invalid" << dendl; + return -EINVAL; } topic_name = topic_arn->resource; @@ -199,21 +204,21 @@ namespace { // ctor and set are done according to the "type" argument // if type is not "key" or "value" its a no-op class Attribute { - std::string key; - std::string value; + std::string key; + std::string value; public: - Attribute(const std::string& type, const std::string& key_or_value) { - set(type, key_or_value); - } - void set(const std::string& type, const std::string& key_or_value) { - if (type == "key") { - key = key_or_value; - } else if (type == "value") { - value = key_or_value; - } + Attribute(const std::string& type, const std::string& key_or_value) { + set(type, key_or_value); + } + void set(const std::string& type, const std::string& key_or_value) { + if (type == "key") { + key = key_or_value; + } else if (type == "value") { + value = key_or_value; } - const std::string& get_key() const { return key; } - const std::string& get_value() const { return value; } + } + const std::string& get_key() const { return key; } + const std::string& get_value() const { return value; } }; using AttributeMap = std::map<unsigned, Attribute>; @@ -431,11 +436,11 @@ void RGWPSCreateNotif_ObjStore_S3::execute() { if (store->getRados()->get_sync_module()) { const auto psmodule = dynamic_cast<RGWPSSyncModuleInstance*>(store->getRados()->get_sync_module().get()); if (psmodule) { - const auto& conf = psmodule->get_effective_conf(); - data_bucket_prefix = conf["data_bucket_prefix"]; - data_oid_prefix = conf["data_oid_prefix"]; - // TODO: allow "push-only" on PS zone as well - push_only = false; + const auto& conf = psmodule->get_effective_conf(); + data_bucket_prefix = conf["data_bucket_prefix"]; + data_oid_prefix = conf["data_oid_prefix"]; + // TODO: allow "push-only" on PS zone as well + push_only = false; } } diff --git a/src/rgw/rgw_rest_pubsub_common.cc b/src/rgw/rgw_rest_pubsub_common.cc index 50f567f7fc1..30d058f7bdb 100644 --- a/src/rgw/rgw_rest_pubsub_common.cc +++ b/src/rgw/rgw_rest_pubsub_common.cc @@ -1,12 +1,50 @@ // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- // vim: ts=8 sw=2 smarttab +#include "rgw_common.h" #include "rgw_rest_pubsub_common.h" #include "common/dout.h" +#include "rgw_url.h" #define dout_context g_ceph_context #define dout_subsys ceph_subsys_rgw +bool validate_and_update_endpoint_secret(rgw_pubsub_sub_dest& dest, CephContext *cct, const RGWEnv& env) { + if (dest.push_endpoint.empty()) { + return true; + } + std::string user; + std::string password; + if (!rgw::parse_url_userinfo(dest.push_endpoint, user, password)) { + ldout(cct, 1) << "endpoint validation error: malformed endpoint URL:" << dest.push_endpoint << dendl; + return false; + } + // this should be verified inside parse_url() + ceph_assert(user.empty() == password.empty()); + if (!user.empty()) { + dest.stored_secret = true; + if (!rgw_transport_is_secure(cct, env)) { + ldout(cct, 1) << "endpoint validation error: sending password over insecure transport" << dendl; + return false; + } + } + return true; +} + +bool subscription_has_endpoint_secret(const rgw_pubsub_sub_config& sub) { + return sub.dest.stored_secret; +} + +bool topic_has_endpoint_secret(const rgw_pubsub_topic_subs& topic) { + return topic.topic.dest.stored_secret; +} + +bool topics_has_endpoint_secret(const rgw_pubsub_user_topics& topics) { + for (const auto& topic : topics.topics) { + if (topic_has_endpoint_secret(topic.second)) return true; + } + return false; +} void RGWPSCreateTopicOp::execute() { op_ret = get_params(); if (op_ret < 0) { @@ -25,10 +63,17 @@ void RGWPSCreateTopicOp::execute() { void RGWPSListTopicsOp::execute() { ups.emplace(store, s->owner.get_id()); op_ret = ups->get_user_topics(&result); + // if there are no topics it is not considered an error + op_ret = op_ret == -ENOENT ? 0 : op_ret; if (op_ret < 0) { ldout(s->cct, 1) << "failed to get topics, ret=" << op_ret << dendl; return; } + if (topics_has_endpoint_secret(result) && !rgw_transport_is_secure(s->cct, *(s->info.env))) { + ldout(s->cct, 1) << "topics contain secret and cannot be sent over insecure transport" << dendl; + op_ret = -EPERM; + return; + } ldout(s->cct, 20) << "successfully got topics" << dendl; } @@ -39,6 +84,11 @@ void RGWPSGetTopicOp::execute() { } ups.emplace(store, s->owner.get_id()); op_ret = ups->get_topic(topic_name, &result); + if (topic_has_endpoint_secret(result) && !rgw_transport_is_secure(s->cct, *(s->info.env))) { + ldout(s->cct, 1) << "topic '" << topic_name << "' contain secret and cannot be sent over insecure transport" << dendl; + op_ret = -EPERM; + return; + } if (op_ret < 0) { ldout(s->cct, 1) << "failed to get topic '" << topic_name << "', ret=" << op_ret << dendl; return; @@ -83,6 +133,11 @@ void RGWPSGetSubOp::execute() { ups.emplace(store, s->owner.get_id()); auto sub = ups->get_sub(sub_name); op_ret = sub->get_conf(&result); + if (subscription_has_endpoint_secret(result) && !rgw_transport_is_secure(s->cct, *(s->info.env))) { + ldout(s->cct, 1) << "subscription '" << sub_name << "' contain secret and cannot be sent over insecure transport" << dendl; + op_ret = -EPERM; + return; + } if (op_ret < 0) { ldout(s->cct, 1) << "failed to get subscription '" << sub_name << "', ret=" << op_ret << dendl; return; @@ -195,7 +250,7 @@ int RGWPSListNotifsOp::verify_permission() { } if (bucket_info.owner != s->owner.get_id()) { - ldout(s->cct, 1) << "user doesn't own bucket, cannot get topic list" << dendl; + ldout(s->cct, 1) << "user doesn't own bucket, cannot get notification list" << dendl; return -EPERM; } diff --git a/src/rgw/rgw_rest_pubsub_common.h b/src/rgw/rgw_rest_pubsub_common.h index 6d78ce5ce1a..f11c75658f5 100644 --- a/src/rgw/rgw_rest_pubsub_common.h +++ b/src/rgw/rgw_rest_pubsub_common.h @@ -6,6 +6,11 @@ #include "rgw_op.h" #include "rgw_pubsub.h" +// make sure that endpoint is a valid URL +// make sure that if user/password are passed inside URL, it is over secure connection +// update rgw_pubsub_sub_dest to indicate that a password is stored in the URL +bool validate_and_update_endpoint_secret(rgw_pubsub_sub_dest& dest, CephContext *cct, const RGWEnv& env); + // create a topic class RGWPSCreateTopicOp : public RGWDefaultResponseOp { protected: diff --git a/src/rgw/rgw_sync_module_pubsub_rest.cc b/src/rgw/rgw_sync_module_pubsub_rest.cc index b198bb33b19..d95b264ea6a 100644 --- a/src/rgw/rgw_sync_module_pubsub_rest.cc +++ b/src/rgw/rgw_sync_module_pubsub_rest.cc @@ -26,6 +26,10 @@ public: topic_name = s->object.name; dest.push_endpoint = s->info.args.get("push-endpoint"); + + if (!validate_and_update_endpoint_secret(dest, s->cct, *(s->info.env))) { + return -EINVAL; + } dest.push_endpoint_args = s->info.args.get_str(); // dest object only stores endpoint info // bucket to store events/records will be set only when subscription is created @@ -169,9 +173,12 @@ public: const auto& conf = psmodule->get_effective_conf(); dest.push_endpoint = s->info.args.get("push-endpoint"); + if (!validate_and_update_endpoint_secret(dest, s->cct, *(s->info.env))) { + return -EINVAL; + } + dest.push_endpoint_args = s->info.args.get_str(); dest.bucket_name = string(conf["data_bucket_prefix"]) + s->owner.get_id().to_str() + "-" + topic_name; dest.oid_prefix = string(conf["data_oid_prefix"]) + sub_name + "/"; - dest.push_endpoint_args = s->info.args.get_str(); dest.arn_topic = topic_name; return 0; diff --git a/src/rgw/rgw_url.cc b/src/rgw/rgw_url.cc new file mode 100644 index 00000000000..24c25378239 --- /dev/null +++ b/src/rgw/rgw_url.cc @@ -0,0 +1,48 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#include <string> +#include <regex> + +namespace rgw { + +namespace { + const auto USER_GROUP_IDX = 3; + const auto PASSWORD_GROUP_IDX = 4; + const auto HOST_GROUP_IDX = 5; + + const std::string schema_re = "([[:alpha:]]+:\\/\\/)"; + const std::string user_pass_re = "(([^:\\s]+):([^@\\s]+)@)?"; + const std::string host_port_re = "([[:alnum:].:-]+)"; +} + +bool parse_url_authority(const std::string& url, std::string& host, std::string& user, std::string& password) { + const std::string re = schema_re + user_pass_re + host_port_re; + const std::regex url_regex(re, std::regex::icase); + std::smatch url_match_result; + + if (std::regex_match(url, url_match_result, url_regex)) { + host = url_match_result[HOST_GROUP_IDX]; + user = url_match_result[USER_GROUP_IDX]; + password = url_match_result[PASSWORD_GROUP_IDX]; + return true; + } + + return false; +} + +bool parse_url_userinfo(const std::string& url, std::string& user, std::string& password) { + const std::string re = schema_re + user_pass_re + host_port_re; + const std::regex url_regex(re); + std::smatch url_match_result; + + if (std::regex_match(url, url_match_result, url_regex)) { + user = url_match_result[USER_GROUP_IDX]; + password = url_match_result[PASSWORD_GROUP_IDX]; + return true; + } + + return false; +} +} + diff --git a/src/rgw/rgw_url.h b/src/rgw/rgw_url.h new file mode 100644 index 00000000000..089401a49a8 --- /dev/null +++ b/src/rgw/rgw_url.h @@ -0,0 +1,12 @@ +// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- +// vim: ts=8 sw=2 smarttab + +#pragma once + +#include <string> +namespace rgw { +// parse a URL of the form: http|https|amqp|amqps|kafka://[user:password@]<host>[:port] +bool parse_url_authority(const std::string& url, std::string& host, std::string& user, std::string& password); +bool parse_url_userinfo(const std::string& url, std::string& user, std::string& password); +} + |