saberbud/rest_rpc
1
0
Fork 0
mirror of https://github.com/qicosmos/rest_rpc.git synced 2026-08-29 16:40:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

support ssl

Browse Source
This commit is contained in:
qicosmos
2020-06-08 09:23:51 +08:00
parent 2594175719
commit 311d5ec5e5
10 changed files with 420 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/connection.h
+154 -25
View File
@@ -8,11 +8,17 @@
#include "use_asio.hpp"
#include "const_vars.h"
#include "router.h"
#include "cplusplus_14.h"
using boost::asio::ip::tcp;
namespace rest_rpc {
namespace rpc_service {
struct ssl_configure {
std::string cert_file;
std::string key_file;
};
class connection : public std::enable_shared_from_this<connection>, private asio::noncopyable {
public:
connection(boost::asio::io_service& io_service, std::size_t timeout_seconds)
@@ -23,9 +29,18 @@ namespace rest_rpc {
has_closed_(false) {
}
~connection() { close(); }
~connection() {
close();
}
void start() { read_head(); }
void start() {
if (is_ssl() && !has_shake_) {
async_handshake();
}
else {
read_head();
}
}
tcp::socket& socket() { return socket_; }
@@ -53,15 +68,6 @@ namespace rest_rpc {
response(req_id, std::move(result));
}
void close() {
has_closed_ = true;
if (socket_.is_open()) {
boost::system::error_code ignored_ec;
socket_.shutdown(tcp::socket::shutdown_both, ignored_ec);
socket_.close(ignored_ec);
}
}
void set_conn_id(int64_t id) { conn_id_ = id; }
int64_t conn_id() const { return conn_id_; }
@@ -87,13 +93,39 @@ namespace rest_rpc {
callback_ = std::move(callback);
}
void init_ssl_context(const ssl_configure& ssl_conf) {
#ifdef CINATRA_ENABLE_SSL
unsigned long ssl_options = boost::asio::ssl::context::default_workarounds
| boost::asio::ssl::context::no_sslv2
| boost::asio::ssl::context::single_dh_use;
try {
boost::asio::ssl::context ssl_context(boost::asio::ssl::context::sslv23);
ssl_context.set_options(ssl_options);
ssl_context.set_password_callback([](std::size_t size,
boost::asio::ssl::context_base::password_purpose purpose) {return "123456"; });
boost::system::error_code ec;
if (fs::exists(ssl_conf.cert_file, ec)) {
ssl_context.use_certificate_chain_file(ssl_conf.cert_file);
}
if (fs::exists(ssl_conf.key_file, ec))
ssl_context.use_private_key_file(ssl_conf.key_file, boost::asio::ssl::context::pem);
//ssl_context_callback(ssl_context);
ssl_stream_ = std::make_unique<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>>(socket_, ssl_context);
}
catch (const std::exception& e) {
print(e);
}
#endif
}
private:
void read_head() {
reset_timer();
auto self(this->shared_from_this());
boost::asio::async_read(
socket_, boost::asio::buffer(head_),
[this, self](boost::system::error_code ec, std::size_t length) {
async_read_head([this, self](boost::system::error_code ec, std::size_t length) {
if (!socket_.is_open()) {
//LOG(INFO) << "socket already closed";
return;
@@ -117,12 +149,12 @@ namespace rest_rpc {
read_head();
}
else {
//LOG(INFO) << "invalid body len";
print("invalid body len");
close();
}
}
else {
//LOG(INFO) << ec.message();
print(ec);
close();
}
});
@@ -130,9 +162,7 @@ namespace rest_rpc {
void read_body(std::size_t size) {
auto self(this->shared_from_this());
boost::asio::async_read(
socket_, boost::asio::buffer(body_.data(), size),
[this, self](boost::system::error_code ec, std::size_t length) {
async_read(size, [this, self](boost::system::error_code ec, std::size_t length) {
cancel_timer();
if (!socket_.is_open()) {
@@ -153,7 +183,7 @@ namespace rest_rpc {
callback_(std::move(std::get<0>(p)), std::move(std::get<1>(p)), this->shared_from_this());
}
catch (const std::exception& ex) {
std::cout << ex.what() << "\n";
print(ex);
}
}
}
@@ -173,8 +203,7 @@ namespace rest_rpc {
write_buffers[3] = boost::asio::buffer(msg.content->data(), write_size_);
auto self = this->shared_from_this();
boost::asio::async_write(
socket_, write_buffers,
async_write(write_buffers,
[this, self](boost::system::error_code ec, std::size_t length) {
on_write(ec, length);
});
@@ -182,8 +211,8 @@ namespace rest_rpc {
void on_write(boost::system::error_code ec, std::size_t length) {
if (ec) {
std::cout << ec.value() << " " << ec.message() << std::endl;
close();
print(ec);
close(false);
return;
}
@@ -197,6 +226,67 @@ namespace rest_rpc {
}
}
void async_handshake() {
#ifdef CINATRA_ENABLE_SSL
auto self = this->shared_from_this();
ssl_stream_->async_handshake(boost::asio::ssl::stream_base::server,
[this, self](const boost::system::error_code& error) {
if (error) {
print(error);
close();
return;
}
has_shake_ = true;
read_head();
});
#endif
}
bool is_ssl() const {
#ifdef CINATRA_ENABLE_SSL
return ssl_stream_ != nullptr;
#else
return false;
#endif
}
template<typename Handler>
void async_read_head(Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
#endif
}
else {
boost::asio::async_read(socket_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
}
}
template<typename Handler>
void async_read(size_t size_to_read, Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
#endif
}
else {
boost::asio::async_read(socket_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
}
}
template<typename BufferType, typename Handler>
void async_write(const BufferType& buffers, Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_write(*ssl_stream_, buffers, std::move(handler));
#endif
}
else {
boost::asio::async_write(socket_, buffers, std::move(handler));
}
}
void reset_timer() {
if (timeout_seconds_ == 0) { return; }
@@ -208,7 +298,7 @@ namespace rest_rpc {
if (ec) { return; }
//LOG(INFO) << "rpc connection timeout";
close();
close(false);
});
}
@@ -218,7 +308,46 @@ namespace rest_rpc {
timer_.cancel();
}
void close(bool close_ssl = true) {
#ifdef CINATRA_ENABLE_SSL
if (close_ssl && ssl_stream_) {
boost::system::error_code ec;
ssl_stream_->shutdown(ec);
ssl_stream_ = nullptr;
}
#endif
if (has_closed_) {
return;
}
boost::system::error_code ignored_ec;
socket_.shutdown(tcp::socket::shutdown_both, ignored_ec);
socket_.close(ignored_ec);
has_closed_ = true;
has_shake_ = false;
}
template<typename... Args>
void print(Args... args) {
#ifdef _DEBUG
std::initializer_list<int>{( std::cout << args << ' ', 0)...};
std::cout << "\n";
#endif
}
void print(const boost::system::error_code& ec) {
print(ec.value(), ec.message());
}
void print(const std::exception& ex) {
print(ex.what());
}
tcp::socket socket_;
#ifdef CINATRA_ENABLE_SSL
std::unique_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>> ssl_stream_ = nullptr;
#endif
bool has_shake_ = false;
char head_[HEAD_LEN];
std::vector<char> body_;
std::uint64_t req_id_;
include/rpc_client.hpp
+141 -34
View File
@@ -52,7 +52,7 @@ namespace rest_rpc {
deadline_(ios_), body_(INIT_BUF_SIZE) {
thd_ = std::make_shared<std::thread>([this] {
ios_.run();
});
});
}
rpc_client(const std::string& host, unsigned short port) : socket_(ios_), work_(ios_),
@@ -62,9 +62,9 @@ namespace rest_rpc {
});
}
~rpc_client() {
~rpc_client() {
close();
stop();
stop();
}
void run(){
@@ -79,22 +79,25 @@ namespace rest_rpc {
reconnect_cnt_ = reconnect_count;
}
bool connect(size_t timeout = 1) {
bool connect(size_t timeout = 3, bool is_ssl = false) {
if (has_connected_)
return true;
assert(port_ != 0);
if (is_ssl) {
upgrade_to_ssl();
}
async_connect();
return wait_conn(timeout);
}
bool connect(const std::string& host, unsigned short port, size_t timeout = 1) {
bool connect(const std::string& host, unsigned short port, bool is_ssl = false, size_t timeout = 3) {
if (port_==0) {
host_ = host;
port_ = port;
}
return connect(timeout);
return connect(timeout, is_ssl);
}
void async_connect(const std::string& host, unsigned short port) {
@@ -137,15 +140,25 @@ namespace rest_rpc {
port_ = port;
}
void close() {
has_connected_ = false;
if (socket_.is_open()) {
boost::system::error_code ignored_ec;
socket_.shutdown(asio::ip::tcp::socket::shutdown_both, ignored_ec);
socket_.close(ignored_ec);
}
clear_cache();
}
void close(bool close_ssl = true) {
boost::system::error_code ec;
if (close_ssl) {
#ifdef CINATRA_ENABLE_SSL
if (ssl_stream_) {
ssl_stream_->shutdown(ec);
ssl_stream_ = nullptr;
}
#endif
}
if (!has_connected_)
return;
has_connected_ = false;
socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
socket_.close(ec);
clear_cache();
}
void set_error_callback(std::function<void(boost::system::error_code)> f) {
err_cb_ = std::move(f);
@@ -237,7 +250,8 @@ namespace rest_rpc {
template<size_t TIMEOUT = DEFAULT_TIMEOUT, typename... Args>
void async_call(const std::string& rpc_name, std::function<void(boost::system::error_code, string_view)> cb, Args&& ... args) {
if (!has_connected_) {
error_callback(boost::asio::error::make_error_code(boost::asio::error::not_connected));
if(cb)
cb(boost::asio::error::make_error_code(boost::asio::error::not_connected), "not connected");
return;
}
@@ -306,6 +320,12 @@ namespace rest_rpc {
call<TIMEOUT>("publish_by_token", std::move(key), std::move(token), std::string(buf.data(), buf.size()));
}
#ifdef CINATRA_ENABLE_SSL
void set_ssl_context_callback(std::function<void(boost::asio::ssl::context&)> ssl_context_callback) {
ssl_context_callback_ = std::move(ssl_context_callback);
}
#endif
private:
void async_connect() {
assert(port_ != 0);
@@ -331,8 +351,13 @@ namespace rest_rpc {
async_reconnect();
}
else {
//std::cout<<"connected ok"<<std::endl;
has_connected_ = true;
//std::cout<<"connected ok"<<std::endl;
if (is_ssl()) {
handshake();
return;
}
has_connected_ = true;
do_read();
resend_subscribe();
if (has_wait_)
@@ -378,17 +403,17 @@ namespace rest_rpc {
void write() {
auto& msg = outbox_[0];
write_size_ = (uint32_t)msg.content.length();
std::array<boost::asio::const_buffer, 4> write_buffers;
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(int32_t));
write_buffers[1] = boost::asio::buffer(&msg.req_id, sizeof(uint64_t));
write_buffers[2] = boost::asio::buffer(&msg.req_type, sizeof(request_type));
write_buffers[3] = boost::asio::buffer((char*)msg.content.data(), write_size_);
std::array<boost::asio::const_buffer, 4> write_buffers;
write_buffers[0] = boost::asio::buffer(&write_size_, sizeof(int32_t));
write_buffers[1] = boost::asio::buffer(&msg.req_id, sizeof(uint64_t));
write_buffers[2] = boost::asio::buffer(&msg.req_type, sizeof(request_type));
write_buffers[3] = boost::asio::buffer((char*)msg.content.data(), write_size_);
boost::asio::async_write(socket_, write_buffers,
async_write(write_buffers,
[this](const boost::system::error_code& ec, const size_t length) {
if (ec) {
has_connected_ = false;
close();
close(false);
error_callback(ec);
return;
@@ -410,8 +435,7 @@ namespace rest_rpc {
}
void do_read() {
boost::asio::async_read(socket_, boost::asio::buffer(head_),
[this](const boost::system::error_code& ec, const size_t length) {
async_read_head([this](const boost::system::error_code& ec, const size_t length) {
if (!socket_.is_open()) {
//LOG(INFO) << "socket already closed";
has_connected_ = false;
@@ -438,17 +462,14 @@ namespace rest_rpc {
}
}
else {
//LOG(INFO) << ec.message();
has_connected_ = false;
close();
close(false);
error_callback(ec);
}
});
}
void read_body(std::uint64_t req_id, request_type req_type, size_t body_len) {
boost::asio::async_read(
socket_, boost::asio::buffer(body_.data(), body_len),
async_read(body_len,
[this, req_id, req_type, body_len](boost::system::error_code ec, std::size_t length) {
//cancel_timer();
@@ -552,9 +573,8 @@ namespace rest_rpc {
it->second(data);
}
catch (const std::exception& ex) {
catch (const std::exception& /*ex*/) {
error_callback(asio::error::make_error_code(asio::error::invalid_argument));
std::cout << ex.what() << "\n";
}
}
@@ -645,8 +665,95 @@ namespace rest_rpc {
};
}
bool is_ssl() const {
#ifdef CINATRA_ENABLE_SSL
return ssl_stream_ != nullptr;
#else
return false;
#endif
}
void handshake() {
#ifdef CINATRA_ENABLE_SSL
ssl_stream_->async_handshake(boost::asio::ssl::stream_base::client,
[this](const boost::system::error_code& ec) {
if (!ec) {
has_connected_ = true;
do_read();
resend_subscribe();
if (has_wait_)
conn_cond_.notify_one();
}
else {
error_callback(ec);
close();
}
});
#endif
}
void upgrade_to_ssl() {
#ifdef CINATRA_ENABLE_SSL
if (ssl_stream_)
return;
boost::asio::ssl::context ssl_context(boost::asio::ssl::context::sslv23);
ssl_context.set_default_verify_paths();
boost::system::error_code ec;
ssl_context.set_options(boost::asio::ssl::context::default_workarounds, ec);
if (ssl_context_callback_) {
ssl_context_callback_(ssl_context);
}
ssl_stream_ = std::make_unique<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>>(socket_, ssl_context);
//verify peer TODO
#else
assert(is_ssl());//please add definition CINATRA_ENABLE_SSL, not allowed coming in this branch
#endif
}
template<typename Handler>
void async_read_head(Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
#endif
}
else {
boost::asio::async_read(socket_, boost::asio::buffer(head_, HEAD_LEN), std::move(handler));
}
}
template<typename Handler>
void async_read(size_t size_to_read, Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_read(*ssl_stream_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
#endif
}
else {
boost::asio::async_read(socket_, boost::asio::buffer(body_.data(), size_to_read), std::move(handler));
}
}
template<typename BufferType, typename Handler>
void async_write(const BufferType& buffers, Handler handler) {
if (is_ssl()) {
#ifdef CINATRA_ENABLE_SSL
boost::asio::async_write(*ssl_stream_, buffers, std::move(handler));
#endif
}
else {
boost::asio::async_write(socket_, buffers, std::move(handler));
}
}
boost::asio::io_service ios_;
asio::ip::tcp::socket socket_;
#ifdef CINATRA_ENABLE_SSL
std::unique_ptr<boost::asio::ssl::stream<boost::asio::ip::tcp::socket&>> ssl_stream_;
std::function<void(boost::asio::ssl::context&)> ssl_context_callback_;
#endif
boost::asio::io_service::work work_;
std::shared_ptr<std::thread> thd_ = nullptr;
include/rpc_server.h
+17 -1
View File
@@ -11,7 +11,7 @@
using boost::asio::ip::tcp;
namespace rest_rpc {
namespace rpc_service {
namespace rpc_service {
using rpc_conn = std::weak_ptr<connection>;
class rpc_server : private asio::noncopyable {
public:
@@ -25,6 +25,15 @@ namespace rest_rpc {
pub_sub_thread_ = std::make_shared<std::thread>([this] { clean_sub_pub(); });
}
rpc_server(short port, size_t size, ssl_configure ssl_conf, size_t timeout_seconds = 15, size_t check_seconds = 10) :
rpc_server(port, size, timeout_seconds, check_seconds) {
#ifdef CINATRA_ENABLE_SSL
ssl_conf_ = std::move(ssl_conf);
#else
assert(false);//please add definition CINATRA_ENABLE_SSL, not allowed coming in this branch
#endif
}
~rpc_server() {
{
std::unique_lock<std::mutex> lock(mtx_);
@@ -99,6 +108,11 @@ namespace rest_rpc {
//LOG(INFO) << "acceptor error: " << ec.message();
}
else {
#ifdef CINATRA_ENABLE_SSL
if (!ssl_conf_.cert_file.empty()) {
conn_->init_ssl_context(ssl_conf_);
}
#endif
conn_->start();
std::unique_lock<std::mutex> lock(mtx_);
conn_->set_conn_id(conn_id_);
@@ -204,6 +218,8 @@ namespace rest_rpc {
std::shared_ptr<std::thread> pub_sub_thread_;
bool stop_check_pub_sub_ = false;
ssl_configure ssl_conf_;
};
} // namespace rpc_service
} // namespace rest_rpc
include/use_asio.hpp
+18
View File
@@ -44,3 +44,21 @@ using namespace nonstd;
using string_view = boost::string_view;
#endif
#endif
#if __cplusplus > 201402L
#if defined (__GNUC__)
#if __GNUC__ < 8
#include <experimental/filesystem>
namespace fs = std::experimental::filesystem;
#else
#include <filesystem>
namespace fs = std::filesystem;
#endif
#else
#include <boost/filesystem.hpp>
namespace fs = boost::filesystem;
#endif
#else
#include <boost/filesystem.hpp>
namespace fs = boost::filesystem;
#endif