Merge pull request #403 from Barenboim/windows

Add MySQL SSL Client (windows)
This commit is contained in:
xiehan
2021-06-03 22:13:11 +08:00
committed by GitHub
11 changed files with 305 additions and 121 deletions
+1 -4
View File
@@ -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()
+234 -76
View File
@@ -22,6 +22,8 @@
#include <string.h>
#include <string>
#include <unordered_map>
#include <openssl/ssl.h>
#include <openssl/bio.h>
#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<MySQLRequest, MySQLResponse>
{
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<handshake_ctx *>(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<RouteManager::RouteTarget *>(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<MySQLHandshakeResponse *>(this->get_message_in());
auto *resp = (MySQLHandshakeResponse *)this->get_message_in();
if (resp->host_disallowed())
{
this->resp = std::move(*static_cast<MySQLResponse *>(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<handshake_ctx *>(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<MySQLResponse *>(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;
}
+1 -1
View File
@@ -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. */
+22 -21
View File
@@ -41,22 +41,6 @@ int WFNetworkTask<REQ, RESP>::get_peer_addr(struct sockaddr *addr,
return -1;
}
template<class REQ, class RESP>
WFConnection *WFNetworkTask<REQ, RESP>::get_connection() const
{
CommConnection *conn;
if (this->target)
{
conn = this->CommSession::get_connection();
if (conn)
return static_cast<WFConnection *>(conn);
}
errno = ENOTCONN;
return NULL;
}
template<class REQ, class RESP>
class WFClientTask : public WFNetworkTask<REQ, RESP>
{
@@ -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<void (WFNetworkTask<REQ, RESP> *)> 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<REQ, RESP>::get_connection();
return (WFConnection *)this->CommSession::get_connection();
errno = EPERM;
return NULL;
+13 -2
View File
@@ -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
+10 -9
View File
@@ -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<HostPort, DNSCacheValue, ValueDeleter> cache_pool_;
};
+3
View File
@@ -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;
}
+12 -4
View File
@@ -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;
}
+7 -3
View File
@@ -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];
+1
View File
@@ -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))
{
+1 -1
View File
@@ -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)
{