Merge pull request #380 from Barenboim/windows

add http proxy code

CMakeLists_Headers.txt

@@ -65,6 +65,7 @@ set(INCLUDE_HEADERS
 	src/protocol/mysql_parser.h
 	src/protocol/mysql_types.h
 	src/protocol/mysql_byteorder.h
+	src/protocol/SSLWrapper.h
 	src/server/WFServer.h
 	src/server/WFHttpServer.h
 	src/server/WFRedisServer.h

src/factory/HttpTaskImpl.cc

@@ -15,15 +15,19 @@
 
   Authors: Wu Jiaxu (wujiaxu@sogou-inc.com)
            Li Yingxin (liyingxin@sogou-inc.com)
+           Liu Kai (liukaidx@sogou-inc.com)
 */
 
 #include <assert.h>
 #include <string>
+#include <openssl/bio.h>
+#include <openssl/ssl.h>
 #include "WFTaskError.h"
 #include "WFTaskFactory.h"
 #include "StringUtil.h"
 #include "WFGlobal.h"
 #include "HttpUtil.h"
+#include "SSLWrapper.h"
 
 using namespace protocol;
 
@@ -56,11 +60,12 @@ protected:
 	virtual void init_failed();
 	virtual bool finish_once();
 
-private:
+protected:
 	bool need_redirect(ParsedURI& uri);
 	bool redirect_url(HttpResponse *client_resp, ParsedURI& uri);
 	void set_empty_request();
 
+private:
 	int redirect_max_;
 	int redirect_count_;
 };
@@ -356,6 +361,395 @@ bool ComplexHttpTask::finish_once()
 	return true;
 }
 
+/*******Proxy Client*******/
+
+static int __encode_auth(const char *p, std::string& auth)
+{
+	static SSL_CTX *init_ssl = WFGlobal::get_ssl_client_ctx();
+	(void)init_ssl;
+	BUF_MEM *bptr;
+	BIO *bmem;
+	BIO *b64;
+
+	b64 = BIO_new(BIO_f_base64());
+	if (b64)
+	{
+		bmem = BIO_new(BIO_s_mem());
+		if (bmem)
+		{
+			BIO_set_flags(b64, BIO_FLAGS_BASE64_NO_NL);
+			b64 = BIO_push(b64, bmem);
+			BIO_write(b64, p, strlen(p));
+			BIO_flush(b64);
+			BIO_get_mem_ptr(b64, &bptr);
+
+			if (bptr->length > 0)
+			{
+				auth.append("Basic ");
+				auth.append(bptr->data, bptr->length);
+			}
+
+			BIO_free_all(b64);
+			return 0;
+		}
+
+		BIO_free_all(b64);
+	}
+
+	return -1;
+}
+
+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;
+}
+
+class ComplexHttpProxyTask : public ComplexHttpTask
+{
+public:
+	ComplexHttpProxyTask(int redirect_max,
+						 int retry_max,
+						 http_callback_t&& callback):
+		ComplexHttpTask(redirect_max, retry_max, std::move(callback))
+	{ }
+
+	void set_user_uri(ParsedURI&& uri) { user_uri_ = std::move(uri); }
+	void set_user_uri(const ParsedURI& uri) { user_uri_ = uri; }
+
+	virtual WFConnection *get_connection() const;
+
+protected:
+	virtual CommMessageOut *message_out();
+	virtual CommMessageIn *message_in();
+	virtual int keep_alive_timeout();
+	virtual bool init_success();
+	virtual bool finish_once();
+
+private:
+	struct ConnContext : public WFConnection
+	{
+		SSL *ssl;
+	};
+
+	SSL *get_ssl() const;
+	int set_ssl();
+
+	std::string proxy_auth_;
+	ParsedURI user_uri_;
+	bool is_ssl_;
+	bool is_user_request_;
+	int state_;
+	int error_;
+};
+
+WFConnection *ComplexHttpProxyTask::get_connection() const
+{
+	WFConnection *conn = this->WFComplexClientTask::get_connection();
+
+	if (conn && is_ssl_)
+		return (ConnContext *)conn->get_context();
+
+	return conn;
+}
+
+SSL *ComplexHttpProxyTask::get_ssl() const
+{
+	WFConnection *conn = this->WFComplexClientTask::get_connection();
+
+	if (conn && is_ssl_)
+		return ((ConnContext *)conn->get_context())->ssl;
+
+	return NULL;
+}
+
+int ComplexHttpProxyTask::set_ssl()
+{
+	SSL *ssl = __create_ssl(WFGlobal::get_ssl_client_ctx());
+	WFConnection *conn;
+
+	if (!ssl)
+		return -1;
+
+	SSL_set_tlsext_host_name(ssl, user_uri_.host);
+	SSL_set_connect_state(ssl);
+
+	conn = this->WFComplexClientTask::get_connection();
+	ConnContext *ctx = new ConnContext;
+	ctx->ssl = ssl;
+
+	auto&& deleter = [] (void *c)
+	{
+		ConnContext *ctx = (ConnContext *)c;
+		SSL_free(ctx->ssl);
+		delete ctx;
+	};
+	conn->set_context(ctx, std::move(deleter));
+	return 0;
+}
+
+CommMessageOut *ComplexHttpProxyTask::message_out()
+{
+	long long seqid = this->get_seq();
+
+	is_user_request_ = false;
+
+	if (seqid == 0) // CONNECT
+	{
+		HttpRequest *conn_req = new HttpRequest;
+		std::string request_uri(user_uri_.host);
+
+		request_uri += ":";
+		if (user_uri_.port)
+			request_uri += user_uri_.port;
+		else
+			request_uri += is_ssl_ ? "443" : "80";
+
+		conn_req->set_method("CONNECT");
+		conn_req->set_request_uri(request_uri);
+		conn_req->set_http_version("HTTP/1.1");
+		conn_req->add_header_pair("Host", request_uri.c_str());
+
+		if (!proxy_auth_.empty())
+			conn_req->add_header_pair("Proxy-Authorization", proxy_auth_);
+
+		return conn_req;
+	}
+	else if (seqid == 1 && is_ssl_) // HANDSHAKE
+		return new SSLHandshaker(this->get_ssl());
+
+	auto *msg = (ProtocolMessage *)this->ComplexHttpTask::message_out();
+	if (is_ssl_)
+		return new SSLWrapper(msg, this->get_ssl());
+
+	is_user_request_ = true;
+	return msg;
+}
+
+CommMessageIn *ComplexHttpProxyTask::message_in()
+{
+	long long seqid = this->get_seq();
+
+	if (seqid == 0)
+	{
+		HttpResponse *conn_resp = new HttpResponse;
+		conn_resp->parse_zero_body();
+		return conn_resp;
+	}
+	else if (seqid == 1 && is_ssl_)
+		return new SSLHandshaker(this->get_ssl());
+
+	auto *msg = (ProtocolMessage *)this->ComplexHttpTask::message_in();
+	if (is_ssl_)
+		return new SSLWrapper(msg, this->get_ssl());
+
+	return msg;
+}
+
+int ComplexHttpProxyTask::keep_alive_timeout()
+{
+	long long seqid = this->get_seq();
+
+	if (seqid == 0)
+	{
+		HttpResponse *resp = this->get_resp();
+		const char *code_str;
+		int status_code;
+
+		*resp = std::move(*(HttpResponse *)this->get_message_in());
+		code_str = resp->get_status_code();
+		status_code = code_str ? atoi(code_str) : 0;
+
+		switch (status_code)
+		{
+		case 200:
+			break;
+		case 407:
+			this->disable_retry();
+		default:
+			state_ = WFT_STATE_TASK_ERROR;
+			error_ = WFT_ERR_HTTP_PROXY_CONNECT_FAILED;
+			return 0;
+		}
+
+		this->clear_resp();
+
+		if (is_ssl_ && set_ssl() < 0)
+		{
+			state_ = WFT_STATE_SYS_ERROR;
+			error_ = errno;
+			return 0;
+		}
+
+		return HTTP_KEEPALIVE_DEFAULT;
+	}
+	else if (seqid == 1 && is_ssl_)
+		return HTTP_KEEPALIVE_DEFAULT;
+
+	return this->ComplexHttpTask::keep_alive_timeout();
+}
+
+bool ComplexHttpProxyTask::init_success()
+{
+	if (!uri_.scheme || strcasecmp(uri_.scheme, "http") != 0)
+	{
+		this->state = WFT_STATE_TASK_ERROR;
+		this->error = WFT_ERR_URI_SCHEME_INVALID;
+		return false;
+	}
+
+	if (user_uri_.state == URI_STATE_ERROR)
+	{
+		this->state = WFT_STATE_SYS_ERROR;
+		this->error = uri_.error;
+		return false;
+	}
+	else if (user_uri_.state != URI_STATE_SUCCESS)
+	{
+		this->state = WFT_STATE_TASK_ERROR;
+		this->error = WFT_ERR_URI_PARSE_FAILED;
+		return false;
+	}
+
+	if (user_uri_.scheme && strcasecmp(user_uri_.scheme, "http") == 0)
+		is_ssl_ = false;
+	else if (user_uri_.scheme && strcasecmp(user_uri_.scheme, "https") == 0)
+		is_ssl_ = true;
+	else
+	{
+		this->state = WFT_STATE_TASK_ERROR;
+		this->error = WFT_ERR_URI_SCHEME_INVALID;
+		this->set_empty_request();
+		return false;
+	}
+
+	int user_port;
+	if (user_uri_.port)
+	{
+		user_port = atoi(user_uri_.port);
+		if (user_port <= 0 || user_port > 65535)
+		{
+			this->state = WFT_STATE_TASK_ERROR;
+			this->error = WFT_ERR_URI_PORT_INVALID;
+			return false;
+		}
+	}
+	else
+		user_port = is_ssl_ ? 443 : 80;
+
+	if (uri_.userinfo && uri_.userinfo[0])
+	{
+		proxy_auth_.clear();
+		if (__encode_auth(uri_.userinfo, proxy_auth_) < 0)
+		{
+			this->state = WFT_STATE_SYS_ERROR;
+			this->error = errno;
+			return false;
+		}
+	}
+
+	std::string info("http-proxy|remote:");
+	info += is_ssl_ ? "https://" : "http://";
+	info += user_uri_.host;
+	info += ":";
+	if (user_uri_.port)
+		info += user_uri_.port;
+	else
+		info += is_ssl_ ? "443" : "80";
+	info += "|auth:";
+	info += proxy_auth_;
+
+	this->WFComplexClientTask::set_info(info);
+
+	std::string request_uri;
+	std::string header_host;
+
+	if (user_uri_.path && user_uri_.path[0])
+		request_uri = user_uri_.path;
+	else
+		request_uri = "/";
+
+	if (user_uri_.query && user_uri_.query[0])
+	{
+		request_uri += "?";
+		request_uri += user_uri_.query;
+	}
+
+	if (user_uri_.host && user_uri_.host[0])
+		header_host = user_uri_.host;
+
+	if ((is_ssl_ && user_port != 443) || (!is_ssl_ && user_port != 80))
+	{
+		header_host += ":";
+		header_host += uri_.port;
+	}
+
+	state_ = WFT_STATE_SUCCESS;
+	error_ = 0;
+	is_user_request_ = true;
+
+	HttpRequest *client_req = this->get_req();
+	client_req->set_request_uri(request_uri.c_str());
+	client_req->set_header_pair("Host", header_host.c_str());
+	this->WFComplexClientTask::set_transport_type(TT_TCP);
+
+	return true;
+}
+
+bool ComplexHttpProxyTask::finish_once()
+{
+	if (!is_user_request_)
+	{
+		long long seqid = this->get_seq();
+
+		delete this->get_message_in();
+		delete this->get_message_out();
+
+		if (this->state == WFT_STATE_SUCCESS && state_ != WFT_STATE_SUCCESS)
+		{
+			this->state = state_;
+			this->error = error_;
+		}
+
+		if (seqid == 0 || (seqid == 1 && is_ssl_))
+			return false;
+	}
+
+	if (this->state == WFT_STATE_SUCCESS)
+	{
+		if (this->need_redirect(user_uri_))
+			this->set_redirect(uri_);
+		else if (this->state != WFT_STATE_SUCCESS)
+			this->disable_retry();
+	}
+	else
+		this->get_resp()->end_parsing();
+
+	return true;
+}
+
 /**********Client Factory**********/
 
 WFHttpTask *WFTaskFactory::create_http_task(const std::string& url,
@@ -388,6 +782,42 @@ WFHttpTask *WFTaskFactory::create_http_task(const ParsedURI& uri,
 	return task;
 }
 
+WFHttpTask *WFTaskFactory::create_http_task(const std::string& url,
+											const std::string& proxy_url,
+											int redirect_max,
+											int retry_max,
+											http_callback_t callback)
+{
+	auto *task = new ComplexHttpProxyTask(redirect_max,
+										  retry_max,
+										  std::move(callback));
+
+	ParsedURI uri, user_uri;
+	URIParser::parse(url, user_uri);
+	URIParser::parse(proxy_url, uri);
+
+	task->set_user_uri(std::move(user_uri));
+	task->set_keep_alive(HTTP_KEEPALIVE_DEFAULT);
+	task->init(std::move(uri));
+	return task;
+}
+
+WFHttpTask *WFTaskFactory::create_http_task(const ParsedURI& uri,
+											const ParsedURI& proxy_uri,
+											int redirect_max,
+											int retry_max,
+											http_callback_t callback)
+{
+	auto *task = new ComplexHttpProxyTask(redirect_max,
+										  retry_max,
+										  std::move(callback));
+
+	task->set_user_uri(uri);
+	task->set_keep_alive(HTTP_KEEPALIVE_DEFAULT);
+	task->init(proxy_uri);
+	return task;
+}
+
 /**********Server**********/
 
 class WFHttpServerTask : public WFServerTask<HttpRequest, HttpResponse>

src/factory/WFTaskError.h

@@ -39,6 +39,7 @@ enum
 
 	//HTTP
 	WFT_ERR_HTTP_BAD_REDIRECT_HEADER = 2001,    ///< Http, 301/302/303/307/308 Location header value is NULL
+	WFT_ERR_HTTP_PROXY_CONNECT_FAILED = 2002,   ///< Http, proxy CONNECT return non 200
 
 	//REDIS
 	WFT_ERR_REDIS_ACCESS_DENIED = 3001,         ///< Redis, invalid password

src/factory/WFTaskFactory.h

@@ -108,6 +108,18 @@ public:
 										int retry_max,
 										http_callback_t callback);
 
+	static WFHttpTask *create_http_task(const std::string& url,
+										const std::string& proxy_url,
+										int redirect_max,
+										int retry_max,
+										http_callback_t callback);
+
+	static WFHttpTask *create_http_task(const ParsedURI& uri,
+										const ParsedURI& proxy_uri,
+										int redirect_max,
+										int retry_max,
+										http_callback_t callback);
+
 	static WFRedisTask *create_redis_task(const std::string& url,
 										  int retry_max,
 										  redis_callback_t callback);

src/manager/WFGlobal.cc

@@ -716,6 +716,9 @@ static inline const char *__get_task_error_string(int error)
 	case WFT_ERR_HTTP_BAD_REDIRECT_HEADER:
 		return "Http Bad Redirect Header";
 
+	case WFT_ERR_HTTP_PROXY_CONNECT_FAILED:
+		return "Http Proxy Connect Failed";
+
 	case WFT_ERR_REDIS_ACCESS_DENIED:
 		return "Redis Access Denied";
 

src/protocol/CMakeLists.txt

@@ -12,6 +12,7 @@ set(SRC
 	HttpMessage.cc
 	RedisMessage.cc
 	HttpUtil.cc
+	SSLWrapper.cc
 )
 
 add_library(${PROJECT_NAME} OBJECT ${SRC})

src/protocol/MySQLMessage.cc

@@ -14,14 +14,18 @@
   limitations under the License.
 
   Authors: Wu Jiaxu (wujiaxu@sogou-inc.com)
+           Xie Han (xiehan@sogou-inc.com)
 */
 
 #include <stdint.h>
 #include <string.h>
 #include <errno.h>
 #include <string>
+#include <openssl/ssl.h>
 #include <openssl/sha.h>
+#include <utility>
 #include "PlatformSocket.h"
+#include "SSLWrapper.h"
 #include "mysql_types.h"
 #include "MySQLResult.h"
 #include "MySQLMessage.h"
@@ -251,6 +255,7 @@ static inline std::string __sha1_bin(const std::string& str)
 	return std::string((const char *)md, 20);
 }
 
+#define MYSQL_CAPFLAG_CLIENT_SSL				0x00000800
 #define MYSQL_CAPFLAG_CLIENT_PROTOCOL_41		0x00000200
 #define MYSQL_CAPFLAG_CLIENT_SECURE_CONNECTION	0x00008000
 #define MYSQL_CAPFLAG_CLIENT_CONNECT_WITH_DB	0x00000008
@@ -259,6 +264,44 @@ static inline std::string __sha1_bin(const std::string& str)
 #define MYSQL_CAPFLAG_CLIENT_PS_MULTI_RESULTS	0x00040000
 #define MYSQL_CAPFLAG_CLIENT_LOCAL_FILES		0x00000080
 
+int MySQLSSLRequest::encode(struct iovec vectors[], int max)
+{
+	unsigned char header[32] = {0};
+	unsigned char *pos = header;
+	int ret;
+
+	int4store(pos, MYSQL_CAPFLAG_CLIENT_SSL |
+				   MYSQL_CAPFLAG_CLIENT_PROTOCOL_41 |
+				   MYSQL_CAPFLAG_CLIENT_SECURE_CONNECTION |
+				   MYSQL_CAPFLAG_CLIENT_CONNECT_WITH_DB |
+				   MYSQL_CAPFLAG_CLIENT_MULTI_RESULTS|
+				   MYSQL_CAPFLAG_CLIENT_LOCAL_FILES |
+				   MYSQL_CAPFLAG_CLIENT_MULTI_STATEMENTS |
+				   MYSQL_CAPFLAG_CLIENT_PS_MULTI_RESULTS);
+	pos += 4;
+	int4store(pos, 0);
+	pos += 4;
+	*pos = (uint8_t)character_set_;
+
+	buf_.clear();
+	buf_.append((char *)header, 32);
+	ret = this->MySQLMessage::encode(vectors, max);
+	if (ret >= 0)
+	{
+		max -= ret;
+		if (max >= 8) /* Indeed SSL handshaker needs only 1 vector. */
+		{
+			max = ssl_handshaker_.encode(vectors + ret, max);
+			if (max >= 0)
+				return max + ret;
+		}
+		else
+			errno = EOVERFLOW;
+	}
+
+	return -1;
+}
+
 int MySQLAuthRequest::encode(struct iovec vectors[], int max)
 {
 	std::string native;

src/protocol/MySQLMessage.inl

@@ -14,11 +14,14 @@
   limitations under the License.
 
   Authors: Wu Jiaxu (wujiaxu@sogou-inc.com)
+           Xie Han (xiehan@sogou-inc.com)
 */
 
 #include <stdint.h>
 #include <string.h>
 #include <string>
+#include <openssl/ssl.h>
+#include "SSLWrapper.h"
 #include "mysql_byteorder.h"
 
 namespace protocol
@@ -77,6 +80,31 @@ public:
 	MySQLHandshakeResponse& operator= (MySQLHandshakeResponse&& move) = default;
 };
 
+class MySQLSSLRequest : public MySQLRequest
+{
+private:
+	virtual int encode(struct iovec vectors[], int max);
+
+	/* Do not support server side with SSL currently. */
+	virtual int decode_packet(const unsigned char *buf, size_t buflen)
+	{
+		return -2;
+	}
+
+private:
+	int character_set_;
+	SSLHandshaker ssl_handshaker_;
+
+public:
+	MySQLSSLRequest(int character_set, SSL *ssl) : ssl_handshaker_(ssl)
+	{
+		character_set_ = character_set;
+	}
+
+	MySQLSSLRequest(MySQLSSLRequest&& move) = default;
+	MySQLSSLRequest& operator= (MySQLSSLRequest&& move) = default;
+};
+
 class MySQLAuthRequest : public MySQLRequest
 {
 public:

src/protocol/ProtocolMessage.h

@@ -33,7 +33,7 @@ namespace protocol
 
 class ProtocolMessage : public CommMessageOut, public CommMessageIn
 {
-private:
+public:
 	virtual int encode(struct iovec vectors[], int max)
 	{
 		errno = ENOSYS;

src/protocol/SSLWrapper.cc

@@ -0,0 +1,262 @@
+/*
+  Copyright (c) 2021 Sogou, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+  Author: Xie Han (xiehan@sogou-inc.com)
+*/
+
+#include <errno.h>
+#include <openssl/ssl.h>
+#include <openssl/err.h>
+#include <openssl/bio.h>
+#include "SSLWrapper.h"
+
+namespace protocol
+{
+
+#if OPENSSL_VERSION_NUMBER < 0x10100000L
+static inline BIO *__get_wbio(SSL *ssl)
+{
+	BIO *wbio = SSL_get_wbio(ssl);
+	BIO *next = BIO_next(wbio);
+	return next ? next : wbio;
+}
+
+# define SSL_get_wbio(ssl)	__get_wbio(ssl)
+#endif
+
+int SSLHandshaker::encode(struct iovec vectors[], int max)
+{
+	BIO *wbio = SSL_get_wbio(this->ssl);
+	char *ptr;
+	long len;
+	int ret;
+
+	if (BIO_reset(wbio) <= 0)
+		return -1;
+
+	ret = SSL_do_handshake(this->ssl);
+	if (ret <= 0)
+	{
+		ret = SSL_get_error(this->ssl, ret);
+		if (ret != SSL_ERROR_WANT_READ)
+		{
+			if (ret != SSL_ERROR_SYSCALL)
+				errno = -ret;
+
+			return -1;
+		}
+	}
+
+	len = BIO_get_mem_data(wbio, &ptr);
+	if (len > 0)
+	{
+		vectors[0].iov_base = ptr;
+		vectors[0].iov_len = len;
+		return 1;
+	}
+	else if (len == 0)
+		return 0;
+	else
+		return -1;
+}
+
+static int __ssl_handshake(const void *buf, size_t *size, SSL *ssl,
+						   char **ptr, long *len)
+{
+	BIO *wbio = SSL_get_wbio(ssl);
+	BIO *rbio = SSL_get_rbio(ssl);
+	int ret;
+
+	if (BIO_reset(wbio) <= 0)
+		return -1;
+
+	ret = BIO_write(rbio, buf, *size);
+	if (ret <= 0)
+		return -1;
+
+	*size = ret;
+	ret = SSL_do_handshake(ssl);
+	if (ret <= 0)
+	{
+		ret = SSL_get_error(ssl, ret);
+		if (ret != SSL_ERROR_WANT_READ)
+		{
+			if (ret != SSL_ERROR_SYSCALL)
+				errno = -ret;
+
+			return -1;
+		}
+
+		ret = 0;
+	}
+
+	*len = BIO_get_mem_data(wbio, ptr);
+	if (*len < 0)
+		return -1;
+
+	return ret;
+}
+
+int SSLHandshaker::append(const void *buf, size_t *size)
+{
+	char *ptr;
+	long len;
+	long n;
+	int ret;
+
+	ret = __ssl_handshake(buf, size, this->ssl, &ptr, &len);
+	if (ret < 0)
+		return -1;
+
+	if (len > 0)
+		n = this->feedback(ptr, len);
+	else
+		n = 0;
+
+	if (n == len)
+		return ret;
+
+	if (n >= 0)
+		errno = EAGAIN;
+
+	return -1;
+}
+
+int SSLWrapper::encode(struct iovec vectors[], int max)
+{
+	BIO *wbio = SSL_get_wbio(this->ssl);
+	struct iovec *iov;
+	char *ptr;
+	long len;
+	int ret;
+
+	if (BIO_reset(wbio) <= 0)
+		return -1;
+
+	ret = this->msg->encode(vectors, max);
+	if ((unsigned int)ret > (unsigned int)max)
+		return ret;
+
+	max = ret;
+	for (iov = vectors; iov < vectors + max; iov++)
+	{
+		if (iov->iov_len > 0)
+		{
+			ret = SSL_write(this->ssl, iov->iov_base, iov->iov_len);
+			if (ret <= 0)
+			{
+				ret = SSL_get_error(this->ssl, ret);
+				if (ret != SSL_ERROR_SYSCALL)
+					errno = -ret;
+
+				return -1;
+			}
+		}
+	}
+
+	len = BIO_get_mem_data(wbio, &ptr);
+	if (len > 0)
+	{
+		vectors[0].iov_base = ptr;
+		vectors[0].iov_len = len;
+		return 1;
+	}
+	else if (len == 0)
+		return 0;
+	else
+		return -1;
+}
+
+#define BUFSIZE		8192
+
+int SSLWrapper::append_message()
+{
+	char buf[BUFSIZE];
+	int ret;
+
+	while ((ret = SSL_read(this->ssl, buf, BUFSIZE)) > 0)
+	{
+		size_t nleft = ret;
+		char *p = buf;
+		size_t n;
+
+		do
+		{
+			n = nleft;
+			ret = this->msg->append(p, &n);
+			if (ret == 0)
+			{
+				nleft -= n;
+				p += n;
+			}
+			else
+				return ret;
+
+		} while (nleft > 0);
+	}
+
+	if (ret < 0)
+	{
+		ret = SSL_get_error(this->ssl, ret);
+		if (ret != SSL_ERROR_WANT_READ)
+		{
+			if (ret != SSL_ERROR_SYSCALL)
+				errno = -ret;
+
+			return -1;
+		}
+	}
+
+	return 0;
+}
+
+int SSLWrapper::append(const void *buf, size_t *size)
+{
+	BIO *rbio = SSL_get_rbio(this->ssl);
+	int ret;
+
+	ret = BIO_write(rbio, buf, *size);
+	if (ret <= 0)
+		return -1;
+
+	*size = ret;
+	return this->append_message();
+}
+
+int ServiceSSLWrapper::append(const void *buf, size_t *size)
+{
+	char *ptr;
+	long len;
+	long n;
+
+	if (__ssl_handshake(buf, size, this->ssl, &ptr, &len) < 0)
+		return -1;
+
+	if (len > 0)
+		n = this->feedback(ptr, len);
+	else
+		n = 0;
+
+	if (n == len)
+		return this->append_message();
+
+	if (n >= 0)
+		errno = EAGAIN;
+
+	return -1;
+}
+
+}
+

src/protocol/SSLWrapper.h

@@ -0,0 +1,88 @@
+/*
+  Copyright (c) 2021 Sogou, Inc.
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+  Author: Xie Han (xiehan@sogou-inc.com)
+*/
+
+#ifndef _SSLWRAPPER_H_
+#define _SSLWRAPPER_H_
+
+#include <openssl/ssl.h>
+#include "ProtocolMessage.h"
+
+namespace protocol
+{
+
+class SSLHandshaker : public ProtocolMessage
+{
+public:
+	virtual int encode(struct iovec vectors[], int max);
+	virtual int append(const void *buf, size_t *size);
+
+protected:
+	SSL *ssl;
+
+public:
+	SSLHandshaker(SSL *ssl) { this->ssl = ssl; }
+
+public:
+	SSLHandshaker(SSLHandshaker&& handshaker) = default;
+	SSLHandshaker& operator = (SSLHandshaker&& handshaker) = default;
+};
+
+class SSLWrapper : public ProtocolMessage
+{
+protected:
+	virtual int encode(struct iovec vectors[], int max);
+	virtual int append(const void *buf, size_t *size);
+
+protected:
+	int append_message();
+
+protected:
+	ProtocolMessage *msg;
+	SSL *ssl;
+
+public:
+	SSLWrapper(ProtocolMessage *msg, SSL *ssl)
+	{
+		this->msg = msg;
+		this->ssl = ssl;
+	}
+
+public:
+	SSLWrapper(SSLWrapper&& wrapper) = default;
+	SSLWrapper& operator = (SSLWrapper&& wrapper) = default;
+};
+
+class ServiceSSLWrapper : public SSLWrapper
+{
+protected:
+	virtual int append(const void *buf, size_t *size);
+
+public:
+	ServiceSSLWrapper(ProtocolMessage *msg, SSL *ssl) : SSLWrapper(msg, ssl)
+	{
+	}
+
+public:
+	ServiceSSLWrapper(ServiceSSLWrapper&& wrapper) = default;
+	ServiceSSLWrapper& operator = (ServiceSSLWrapper&& wrapper) = default;
+};
+
+}
+
+#endif
+