diff --git a/src/factory/HttpTaskImpl.cc b/src/factory/HttpTaskImpl.cc index 51073319..11ab9624 100644 --- a/src/factory/HttpTaskImpl.cc +++ b/src/factory/HttpTaskImpl.cc @@ -551,10 +551,7 @@ CommMessageOut *ComplexHttpProxyTask::message_out() } auto *msg = (ProtocolMessage *)this->ComplexHttpTask::message_out(); - if (is_ssl_) - return get_ssl_wrapper(msg); - - return msg; + return is_ssl_ ? get_ssl_wrapper(msg) : msg; } CommMessageIn *ComplexHttpProxyTask::message_in() diff --git a/src/factory/MySQLTaskImpl.cc b/src/factory/MySQLTaskImpl.cc index 641a8293..12fa333d 100644 --- a/src/factory/MySQLTaskImpl.cc +++ b/src/factory/MySQLTaskImpl.cc @@ -22,6 +22,8 @@ #include #include #include +#include +#include #include "WFTaskError.h" #include "WFTaskFactory.h" #include "MySQLMessage.h" @@ -35,12 +37,6 @@ using namespace protocol; /**********Client**********/ -struct handshake_ctx -{ - char challenge[20]; - unsigned char mysql_seqid; -}; - class ComplexMySQLTask : public WFComplexClientTask { protected: @@ -51,6 +47,58 @@ protected: virtual bool init_success(); virtual bool finish_once(); +protected: + virtual WFConnection *get_connection() const + { + WFConnection *conn = this->WFComplexClientTask::get_connection(); + + if (conn && is_ssl_) + return (SSLConnection *)conn->get_context(); + + return conn; + } + +private: + struct SSLConnection : public WFConnection + { + SSL *ssl_; + SSLWrapper wrapper_; + SSLConnection(SSL *ssl) : wrapper_(NULL, ssl) + { + ssl_ = ssl; + } + }; + + SSL *get_ssl() const + { + return ((SSLConnection *)this->get_connection())->ssl_; + } + + SSLWrapper *get_ssl_wrapper(ProtocolMessage *msg) const + { + SSLConnection *conn = (SSLConnection *)this->get_connection(); + conn->wrapper_ = SSLWrapper(msg, conn->ssl_); + return &conn->wrapper_; + } + + int init_ssl_connection(); + + struct MySSLWrapper : public SSLWrapper + { + MySSLWrapper(ProtocolMessage *msg, SSL *ssl) : + SSLWrapper(msg, ssl) + { } + ProtocolMessage *get_msg() const { return this->msg; } + virtual ~MySSLWrapper() { delete this->msg; } + }; + +private: + struct handshake_ctx + { + char challenge[20]; + unsigned char mysql_seqid; + }; + private: std::string username_; std::string password_; @@ -61,9 +109,11 @@ private: #define TRANSACTION_OUT 0 #define TRANSACTION_IN 1 #define TRANSACTION_CONN_RESET -2 - int transaction_state_; + short transaction_state_; #define PREPARE_IN 2 - bool succ_; + short state_; + int error_; + bool is_ssl_; bool is_user_request_; public: @@ -124,44 +174,110 @@ bool ComplexMySQLTask::check_request() return false; } +static SSL *__create_ssl(SSL_CTX *ssl_ctx) +{ + BIO *wbio; + BIO *rbio; + SSL *ssl; + + rbio = BIO_new(BIO_s_mem()); + if (rbio) + { + wbio = BIO_new(BIO_s_mem()); + if (wbio) + { + ssl = SSL_new(ssl_ctx); + if (ssl) + { + SSL_set_bio(ssl, rbio, wbio); + return ssl; + } + + BIO_free(wbio); + } + + BIO_free(rbio); + } + + return NULL; +} + +int ComplexMySQLTask::init_ssl_connection() +{ + SSL *ssl = __create_ssl(WFGlobal::get_ssl_client_ctx()); + WFConnection *conn; + + if (!ssl) + return -1; + + SSL_set_connect_state(ssl); + + conn = this->WFComplexClientTask::get_connection(); + SSLConnection *ssl_conn = new SSLConnection(ssl); + + auto&& deleter = [] (void *ctx) + { + SSLConnection *ssl_conn = (SSLConnection *)ctx; + SSL_free(ssl_conn->ssl_); + delete ssl_conn; + }; + conn->set_context(ssl_conn, std::move(deleter)); + return 0; +} + CommMessageOut *ComplexMySQLTask::message_out() { long long seqid = this->get_seq(); - MySQLRequest *req; + is_user_request_ = false; if (seqid == 0) - req = new MySQLHandshakeRequest; - else if (seqid == 1) - { - auto *auth_req = new MySQLAuthRequest; - auto *conn = this->get_connection(); - auto *ctx = static_cast(conn->get_context()); + return new MySQLHandshakeRequest; - auth_req->set_seqid(ctx->mysql_seqid); - auth_req->set_challenge(ctx->challenge); + if (is_ssl_) + { + if (seqid == 1) + { + auto *req = new MySQLSSLRequest(character_set_, get_ssl()); + auto *conn = this->get_connection(); + auto *ctx = (struct handshake_ctx *)conn->get_context(); + + req->set_seqid(ctx->mysql_seqid++); + return req; + } + + seqid--; + } + + if (seqid == 1) + { + auto *req = new MySQLAuthRequest; + auto *conn = this->get_connection(); + auto *ctx = (struct handshake_ctx *)conn->get_context(); + + req->set_seqid(ctx->mysql_seqid++); + req->set_challenge(ctx->challenge); delete ctx; conn->set_context(NULL, nullptr); - auth_req->set_auth(username_, password_, db_, character_set_); - req = auth_req; + req->set_auth(username_, password_, db_, character_set_); + if (is_ssl_) + return new MySSLWrapper(req, get_ssl()); + else + return req; } else if (seqid == 2 && res_charset_.size() != 0) { - req = new MySQLRequest; + auto *req = new MySQLRequest; req->set_query("SET NAMES " + res_charset_); - } - else - req = NULL; - - if (req) - { - succ_ = false; - is_user_request_ = false; - return req; + if (is_ssl_) + return new MySSLWrapper(req, get_ssl()); + else + return req; } + is_user_request_ = true; if (is_transaction()) { - auto *target = static_cast(this->get_target()); + auto *target = (RouteManager::RouteTarget *)this->get_target(); if (seqid <= 3 && (seqid == 2 || res_charset_.size() != 0) && (target->state & (TRANSACTION_IN | PREPARE_IN))) @@ -210,60 +326,112 @@ CommMessageOut *ComplexMySQLTask::message_out() } } - return this->WFClientTask::message_out(); + auto *msg = (ProtocolMessage *)this->WFComplexClientTask::message_out(); + return is_ssl_ ? get_ssl_wrapper(msg) : msg; } CommMessageIn *ComplexMySQLTask::message_in() { long long seqid = this->get_seq(); + ProtocolMessage *resp; if (seqid == 0) return new MySQLHandshakeResponse; - else if (seqid == 1) - return new MySQLAuthResponse; - else if (seqid == 2 && !is_user_request_) - return new MySQLResponse; - return this->WFClientTask::message_in(); + if (is_ssl_) + { + if (seqid == 1) + return new SSLHandshaker(get_ssl()); + + seqid--; + } + + if (seqid == 1) + resp = new MySQLAuthResponse; + else if (seqid == 2 && !is_user_request_) + resp = new MySQLResponse; + else + resp = (ProtocolMessage *)this->WFClientTask::message_in(); + + if (!is_ssl_) + return resp; + + if (is_user_request_) + return get_ssl_wrapper(resp); + else + return new MySSLWrapper(resp, get_ssl()); } int ComplexMySQLTask::keep_alive_timeout() { long long seqid = this->get_seq(); + state_ = WFT_STATE_SUCCESS; + error_ = 0; if (seqid == 0) { - auto *resp = static_cast(this->get_message_in()); + auto *resp = (MySQLHandshakeResponse *)this->get_message_in(); if (resp->host_disallowed()) { - this->resp = std::move(*static_cast(resp)); - succ_ = false; + this->resp = std::move(*(MySQLResponse *)resp); + state_ = WFT_STATE_TASK_ERROR; + error_ = WFT_ERR_MYSQL_HOST_NOT_ALLOWED; return 0; } - else + + if (is_ssl_) { - auto *ctx = new handshake_ctx(); - auto *conn = this->get_connection(); + if (!(resp->get_capability_flags() & 0x800)) + { + this->resp = std::move(*(MySQLResponse *)resp); + state_ = WFT_STATE_TASK_ERROR; + error_ = WFT_ERR_MYSQL_SSL_NOT_SUPPORTED; + return 0; + } - ctx->mysql_seqid = resp->get_seqid() + 1; - resp->get_challenge(ctx->challenge); - conn->set_context(ctx, [](void *ctx) { - delete static_cast(ctx); - }); - - succ_ = true; + if (init_ssl_connection() < 0) + { + state_ = WFT_STATE_SYS_ERROR; + error_ = errno; + return 0; + } } + + auto *ctx = new handshake_ctx(); + auto *conn = this->get_connection(); + + ctx->mysql_seqid = resp->get_seqid() + 1; + resp->get_challenge(ctx->challenge); + conn->set_context(ctx, [](void *ctx) { + delete (handshake_ctx *)ctx; + }); } else if (!is_user_request_) { - auto *resp = static_cast(this->get_message_in()); - - succ_ = resp->is_ok_packet(); - if (!succ_) + if (!is_ssl_ || seqid != 1) { - this->resp = std::move(*resp); - return 0; + auto *msg = (ProtocolMessage *)this->get_message_in(); + MySQLResponse *resp; + + if (is_ssl_) + resp = (MySQLResponse *)((MySSLWrapper *)msg)->get_msg(); + else + resp = (MySQLResponse *)msg; + + if (!resp->is_ok_packet()) + { + this->resp = std::move(*resp); + + if (is_ssl_) + seqid--; + + state_ = WFT_STATE_TASK_ERROR; + if (seqid == 1) + error_ = WFT_ERR_MYSQL_ACCESS_DENIED; + else + error_ = WFT_ERR_MYSQL_INVALID_CHARACTER_SET; + } } } else @@ -376,12 +544,10 @@ static int __mysql_get_character_set(const std::string& charset) bool ComplexMySQLTask::init_success() { - TransportType type; - if (uri_.scheme && strcasecmp(uri_.scheme, "mysql") == 0) - type = TT_TCP; - //else if (uri_.scheme && strcasecmp(uri_.scheme, "mysql_ssl") == 0) - // type = TT_TCP_SSL; + is_ssl_ = false; + else if (uri_.scheme && strcasecmp(uri_.scheme, "mysqls") == 0) + is_ssl_ = true; else { this->state = WFT_STATE_TASK_ERROR; @@ -460,11 +626,11 @@ bool ComplexMySQLTask::init_success() res_charset_.size() + 50; char *info = new char[info_len]; - snprintf(info, info_len, "mysql|user:%s|pass:%s|db:%s|" + snprintf(info, info_len, "%s|user:%s|pass:%s|db:%s|" "charset:%d|rcharset:%s", - username_.c_str(), password_.c_str(), db_.c_str(), - character_set_, res_charset_.c_str()); - this->WFComplexClientTask::set_transport_type(type); + is_ssl_ ? "mysqls" : "mysql", username_.c_str(), password_.c_str(), + db_.c_str(), character_set_, res_charset_.c_str()); + this->WFComplexClientTask::set_transport_type(TT_TCP); if (!transaction.empty()) { @@ -488,25 +654,17 @@ bool ComplexMySQLTask::finish_once() { if (!is_user_request_) { - is_user_request_ = true; delete this->get_message_out(); delete this->get_message_in(); - if (this->state == WFT_STATE_SUCCESS && !succ_) + if (this->state == WFT_STATE_SUCCESS && state_ != WFT_STATE_SUCCESS) { - long long seqid = this->get_seq(); - - if (seqid == 0) - this->error = WFT_ERR_MYSQL_HOST_NOT_ALLOWED; - else if (seqid == 1) - this->error = WFT_ERR_MYSQL_ACCESS_DENIED; - else - this->error = WFT_ERR_MYSQL_INVALID_CHARACTER_SET; - + this->state = state_; + this->error = error_; this->disable_retry(); - this->state = WFT_STATE_TASK_ERROR; } + is_user_request_ = true; return false; } diff --git a/src/factory/WFTask.h b/src/factory/WFTask.h index f4b17956..2f715ff1 100644 --- a/src/factory/WFTask.h +++ b/src/factory/WFTask.h @@ -260,7 +260,7 @@ public: int get_peer_addr(struct sockaddr *addr, socklen_t *addrlen) const; - virtual WFConnection *get_connection() const; + virtual WFConnection *get_connection() const = 0; public: /* All in milliseconds. timeout == -1 for unlimited. */ diff --git a/src/factory/WFTask.inl b/src/factory/WFTask.inl index 089e6e85..941fbd65 100644 --- a/src/factory/WFTask.inl +++ b/src/factory/WFTask.inl @@ -41,22 +41,6 @@ int WFNetworkTask::get_peer_addr(struct sockaddr *addr, return -1; } -template -WFConnection *WFNetworkTask::get_connection() const -{ - CommConnection *conn; - - if (this->target) - { - conn = this->CommSession::get_connection(); - if (conn) - return static_cast(conn); - } - - errno = ENOTCONN; - return NULL; -} - template class WFClientTask : public WFNetworkTask { @@ -73,6 +57,22 @@ protected: virtual CommMessageIn *message_in() { return &this->resp; } +protected: + virtual WFConnection *get_connection() const + { + CommConnection *conn; + + if (this->target) + { + conn = this->CommSession::get_connection(); + if (conn) + return (WFConnection *)conn; + } + + errno = ENOTCONN; + return NULL; + } + public: void set_prepare(std::function *)> prep) { @@ -106,7 +106,7 @@ protected: { if (this->state == WFT_STATE_TOREPLY) { - /* After reply success, get_connection() is enabled again. */ + /* Enable get_connection() again if the reply() call is success. */ this->processor.task = this; if (this->scheduler->reply(this) >= 0) return; @@ -119,13 +119,14 @@ protected: this->subtask_done(); } - /* CommSession's get_connection() is supposed to be called only in its - * virtual functions. As a server task, call of this method after process() - * and before callback() is very dangerous and should be blocked. */ + /* CommSession::get_connection() is supposed to be called only in the + * implementations of it's virtual functions. As a server task, to call + * this function after process() and before callback() is very dangerous + * and should be blocked. */ virtual WFConnection *get_connection() const { if (this->processor.task) - return this->WFNetworkTask::get_connection(); + return (WFConnection *)this->CommSession::get_connection(); errno = EPERM; return NULL; diff --git a/src/factory/WFTaskError.h b/src/factory/WFTaskError.h index 9d491454..3658c86c 100644 --- a/src/factory/WFTaskError.h +++ b/src/factory/WFTaskError.h @@ -35,7 +35,6 @@ enum WFT_ERR_URI_SCHEME_INVALID = 1002, ///< URI, invalid scheme WFT_ERR_URI_PORT_INVALID = 1003, ///< URI, invalid port WFT_ERR_UPSTREAM_UNAVAILABLE = 1004, ///< Upstream, all target server down - WFT_ERR_ROUTE_FAILED = 1005, ///< Route, route task is NULL //HTTP WFT_ERR_HTTP_BAD_REDIRECT_HEADER = 2001, ///< Http, 301/302/303/307/308 Location header value is NULL @@ -50,7 +49,19 @@ enum WFT_ERR_MYSQL_ACCESS_DENIED = 4002, ///< MySQL, authentication failed WFT_ERR_MYSQL_INVALID_CHARACTER_SET = 4003, ///< MySQL, invalid charset, not found in MySQL-Documentation WFT_ERR_MYSQL_COMMAND_DISALLOWED = 4004, ///< MySQL, sql command disabled, cannot be "USE"/"SET NAMES"/"SET CHARSET"/"SET CHARACTER SET" - WFT_ERR_MYSQL_QUERY_NOT_SET = 4005 ///< MySQL, query not set sql, maybe forget please check + WFT_ERR_MYSQL_QUERY_NOT_SET = 4005, ///< MySQL, query not set sql, maybe forget please check + WFT_ERR_MYSQL_SSL_NOT_SUPPORTED = 4006, ///< MySQL, SSL not supported by the server + + //KAFKA + WFT_ERR_KAFKA_PARSE_RESPONSE_FAILED = 5001, ///< Kafka parse response failed + WFT_ERR_KAFKA_PRODUCE_FAILED = 5002, + WFT_ERR_KAFKA_FETCH_FAILED = 5003, + WFT_ERR_KAFKA_CGROUP_FAILED = 5004, + WFT_ERR_KAFKA_COMMIT_FAILED = 5005, + WFT_ERR_KAFKA_META_FAILED = 5006, + WFT_ERR_KAFKA_LEAVEGROUP_FAILED = 5007, + WFT_ERR_KAFKA_API_UNKNOWN = 5008, ///< api type not supported + WFT_ERR_KAFKA_VERSION_DISALLOWED = 5009, ///< broker version not supported }; #endif diff --git a/src/manager/DNSCache.h b/src/manager/DNSCache.h index 1b9a29ed..6c097c5c 100644 --- a/src/manager/DNSCache.h +++ b/src/manager/DNSCache.h @@ -36,15 +36,6 @@ struct DNSCacheValue int64_t expire_time; }; -class ValueDeleter -{ -public: - void operator() (const DNSCacheValue& value) const - { - freeaddrinfo(value.addrinfo); - } -}; - // RAII: NO. Release handle by user // Thread safety: YES // MUST call release when handle no longer used @@ -157,6 +148,16 @@ private: const DNSHandle *get_inner(const HostPort& host_port, int type); std::mutex mutex_; + + class ValueDeleter + { + public: + void operator() (const DNSCacheValue& value) const + { + freeaddrinfo(value.addrinfo); + } + }; + LRUCache cache_pool_; }; diff --git a/src/manager/WFGlobal.cc b/src/manager/WFGlobal.cc index ac18556a..16afa73f 100644 --- a/src/manager/WFGlobal.cc +++ b/src/manager/WFGlobal.cc @@ -737,6 +737,9 @@ static inline const char *__get_task_error_string(int error) case WFT_ERR_MYSQL_COMMAND_DISALLOWED: return "MySQL Command Disallowed"; + case WFT_ERR_MYSQL_SSL_NOT_SUPPORTED: + return "MySQL SSL Not Supported"; + default: break; } diff --git a/src/protocol/MySQLMessage.cc b/src/protocol/MySQLMessage.cc index 0b7954e9..aa83af4c 100644 --- a/src/protocol/MySQLMessage.cc +++ b/src/protocol/MySQLMessage.cc @@ -176,18 +176,21 @@ std::string MySQLRequest::get_query() const int MySQLHandshakeResponse::encode(struct iovec vectors[], int max) { - const char empty13[13] = {0}; + const char empty[11] = {0}; + uint16_t cap_flags_lower = capability_flags_ & 0xffffffff; + uint16_t cap_flags_upper = capability_flags_ >> 16; buf_.clear(); buf_.append((const char *)&protocol_version_, 1); buf_.append(server_version_.c_str(), server_version_.size() + 1); buf_.append((const char *)&connection_id_, 4); buf_.append((const char *)auth_plugin_data_part_1_, 8); - buf_.append(empty13, 1); - buf_.append(empty13, 2); + buf_.append(empty, 1); + buf_.append((const char *)&cap_flags_lower, 2); buf_.append((const char *)&character_set_, 1); buf_.append((const char *)&status_flags_, 2); - buf_.append(empty13, 13); + buf_.append((const char *)&cap_flags_upper, 2); + buf_.append(empty, 11); buf_.append((const char *)auth_plugin_data_part_2_, 12); return this->MySQLMessage::encode(vectors, max); } @@ -196,6 +199,8 @@ int MySQLHandshakeResponse::decode_packet(const unsigned char *buf, size_t bufle { const unsigned char *end = buf + buflen; const unsigned char *pos; + uint16_t cap_flags_lower; + uint16_t cap_flags_upper; if (buflen == 0) return -2; @@ -230,12 +235,15 @@ int MySQLHandshakeResponse::decode_packet(const unsigned char *buf, size_t bufle buf += 4; memcpy(auth_plugin_data_part_1_, buf, 8); buf += 9; + cap_flags_lower = uint2korr(buf); buf += 2; character_set_ = *buf++; status_flags_ = uint2korr(buf); buf += 2; + cap_flags_upper = uint2korr(buf); buf += 13; memcpy(auth_plugin_data_part_2_, buf, 12); + capability_flags_ = (cap_flags_upper << 16U) + cap_flags_lower; return 1; } diff --git a/src/protocol/MySQLMessage.inl b/src/protocol/MySQLMessage.inl index fde9a2ae..166c891c 100644 --- a/src/protocol/MySQLMessage.inl +++ b/src/protocol/MySQLMessage.inl @@ -46,19 +46,22 @@ public: void server_set(uint8_t protocol_version, const std::string server_version, uint32_t connection_id, const uint8_t *auth1, - uint8_t character_set, uint16_t status_flags, - const uint8_t *auth2) + uint32_t capability_flags, uint8_t character_set, + uint16_t status_flags, const uint8_t *auth2) { protocol_version_ = protocol_version; server_version_ = server_version; connection_id_ = connection_id; memcpy(auth_plugin_data_part_1_, auth1, 8); + capability_flags_ = capability_flags; character_set_ = character_set; status_flags_ = status_flags; memcpy(auth_plugin_data_part_2_, auth2, 12); } - bool host_disallowed() const { return this->disallowed_; } + bool host_disallowed() const { return disallowed_; } + uint32_t get_capability_flags() const { return capability_flags_; } + uint16_t get_status_flags() const { return status_flags_; } private: virtual int decode_packet(const unsigned char *buf, size_t buflen); @@ -67,6 +70,7 @@ private: std::string server_version_; uint32_t connection_id_; uint8_t auth_plugin_data_part_1_[8]; + uint32_t capability_flags_; uint8_t character_set_; uint16_t status_flags_; uint8_t auth_plugin_data_part_2_[12]; diff --git a/src/protocol/MySQLResult.cc b/src/protocol/MySQLResult.cc index ad3727a4..4764f9d2 100644 --- a/src/protocol/MySQLResult.cc +++ b/src/protocol/MySQLResult.cc @@ -131,6 +131,7 @@ void MySQLResultCursor::init(const MySQLResponse *resp) this->field_count = 0; this->fields = NULL; this->parser = resp->get_parser(); + this->status = MYSQL_STATUS_NOT_INIT; if (!list_empty(&this->parser->result_set_list)) { diff --git a/src/server/WFMySQLServer.cc b/src/server/WFMySQLServer.cc index 07be8ad3..95c7f702 100644 --- a/src/server/WFMySQLServer.cc +++ b/src/server/WFMySQLServer.cc @@ -33,7 +33,7 @@ WFConnection *WFMySQLServer::new_connection(int accept_fd) int count; resp.server_set(0x0a, "5.5", 1, (const uint8_t *)"12345678", - 33, 0, (const uint8_t *)"123456789abc"); + 0, 33, 0, (const uint8_t *)"123456789abc"); count = resp.encode(vec, 8); if (count >= 0) {