Peerplays_Blockchain/peerplays-fc
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Updates from BitShares FC #22

Closed
nathanielhourt wants to merge 693 commits from dapp-support into latest-fc
pull from: dapp-support
merge into: Peerplays_Blockchain:latest-fc
Conversation 4 Commits 693 Files changed 313 +78 -72
2 changed files with 78 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit ba394e05b3 - Show all commits

14
include/fc/network/http/websocket.hpp
View file

@ -31,9 +31,12 @@ namespace fc { namespace http {
boost::any& get_session_data() { return _session_data; }
virtual std::string get_request_header(const std::string& key) = 0;
virtual std::string get_remote_ip(const std::string& forward_header_key) = 0;
const std::string& get_remote_endpoint()const { return _remote_endpoint; }
fc::signal<void()> closed;
protected:
std::string _remote_endpoint; // for logging
private:
boost::any _session_data;
std::function<void(const std::string&)> _on_message;
@ -46,7 +49,7 @@ namespace fc { namespace http {
class websocket_server
{
public:
websocket_server();
websocket_server(const std::string& forward_header_key);
~websocket_server();
void on_connection( const on_connection_handler& handler);
@ -54,7 +57,6 @@ namespace fc { namespace http {
void listen( const fc::ip::endpoint& ep );
uint16_t get_listening_port();
void start_accept();
void set_forward_header_key(const std::string& key);
void stop_listening();
void close();
@ -68,15 +70,15 @@ namespace fc { namespace http {
class websocket_tls_server
{
public:
websocket_tls_server( const std::string& server_pem = std::string(),
const std::string& ssl_password = std::string());
websocket_tls_server( const std::string& server_pem,
const std::string& ssl_password,
const std::string& forward_header_key );
~websocket_tls_server();
void on_connection( const on_connection_handler& handler);
void listen( uint16_t port );
void listen( const fc::ip::endpoint& ep );
void start_accept();
void set_forward_header_key(const std::string& key);
private:
friend class detail::websocket_tls_server_impl;
std::unique_ptr<detail::websocket_tls_server_impl> my;

136
src/network/http/websocket.cpp
View file

@ -20,6 +20,8 @@
#include <fc/thread/thread.hpp>
#include <fc/asio.hpp>
#include <boost/algorithm/string.hpp>
#ifdef DEFAULT_LOGGER
# undef DEFAULT_LOGGER
#endif
@ -144,8 +146,14 @@ namespace fc { namespace http {
class websocket_connection_impl : public websocket_connection
{
public:
websocket_connection_impl( T con )
:_ws_connection(con){
websocket_connection_impl( T con, const std::string& forward_header_key ) : _ws_connection(con)
{
// Firstly, try to extract remote address from a XFF-like header
if( !forward_header_key.empty() )
_remote_endpoint = get_request_header( forward_header_key );
// If the last step failed, retrieve the remote address from the connection
if( _remote_endpoint.empty() )
_remote_endpoint = con->get_remote_endpoint();
}
virtual ~websocket_connection_impl()
@ -154,7 +162,8 @@ namespace fc { namespace http {
virtual void send_message( const std::string& message )override
{
idump((message));
ilog( "[OUT] ${remote_endpoint} ${msg}",
("remote_endpoint",_remote_endpoint) ("msg",message) );
auto ec = _ws_connection->send( message );
FC_ASSERT( !ec, "websocket send failed: ${msg}", ("msg",ec.message() ) );
}
@ -168,30 +177,7 @@ namespace fc { namespace http {
return _ws_connection->get_request_header(key);
}
/****
* @brief retrieves the remote IP address
*
* @param forward_header_key the key to look at in the request header
* @returns the value in the header, otherwise the remote endpoint
*/
virtual std::string get_remote_ip(const std::string& forward_header_key) override
{
if (last_ip.empty() || last_forward_header_key != forward_header_key)
{
// refresh the cache
last_forward_header_key = forward_header_key;
if (!forward_header_key.empty())
last_ip = get_request_header(forward_header_key);
if (last_ip.empty())
last_ip = _ws_connection->get_remote_endpoint();
}
return last_ip;
}
T _ws_connection;
// cache the value of the remote IP
std::string last_forward_header_key;
std::string last_ip;
};
typedef websocketpp::lib::shared_ptr<boost::asio::ssl::context> context_ptr;
@ -199,8 +185,8 @@ namespace fc { namespace http {
class websocket_server_impl
{
public:
websocket_server_impl()
:_server_thread( fc::thread::current() ), fwd_header_key("")
websocket_server_impl( const std::string& forward_header_key )
:_server_thread( fc::thread::current() ), _forward_header_key(forward_header_key)
{
_server.clear_access_channels( websocketpp::log::alevel::all );
_server.init_asio(&fc::asio::default_io_service());
@ -208,7 +194,8 @@ namespace fc { namespace http {
_server.set_open_handler( [&]( connection_hdl hdl ){
_server_thread.async( [this, hdl](){
auto new_con = std::make_shared<websocket_connection_impl<
websocket_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl) );
websocket_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl),
_forward_header_key );
_on_connection( _connections[hdl] = new_con );
}).wait();
});
@ -218,11 +205,15 @@ namespace fc { namespace http {
assert( current_con != _connections.end() );
auto payload = msg->get_payload();
std::shared_ptr<websocket_connection> con = current_con->second;
wlog( "Websocket Server Remote: ${host} Payload: ${body}",
("host", con->get_remote_ip(fwd_header_key)) ("body", msg->get_payload()));
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",con->get_remote_endpoint()) ("msg",payload) );
++_pending_messages;
auto f = fc::async([this,con,payload](){ if( _pending_messages ) --_pending_messages; con->on_message( payload ); });
if( _pending_messages > 100 )
auto f = fc::async([this,con,payload](){
if( _pending_messages )
--_pending_messages;
con->on_message( payload );
});
if( _pending_messages > 100 )
f.wait();
}).wait();
});
@ -234,17 +225,21 @@ namespace fc { namespace http {
_server.set_http_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
auto current_con = std::make_shared<websocket_connection_impl<websocket_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl) );
auto con = _server.get_con_from_hdl(hdl);
auto current_con = std::make_shared<websocket_connection_impl<
websocket_server_type::connection_ptr>>( con, _forward_header_key );
_on_connection( current_con );
auto con = _server.get_con_from_hdl(hdl);
con->defer_http_response();
std::string remote_endpoint = current_con->get_remote_endpoint();
std::string request_body = con->get_request_body();
wdump(("server")(request_body));
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",request_body) );
fc::async([current_con, request_body, con] {
fc::async([current_con, request_body, con, remote_endpoint] {
fc::http::reply response = current_con->on_http(request_body);
idump( (response) );
ilog( "[OUT] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",response) );
con->set_body( std::move( response.body_as_string ) );
con->set_status( websocketpp::http::status_code::value(response.status) );
con->send_http_response();
@ -303,8 +298,6 @@ namespace fc { namespace http {
if( _closed ) _closed->wait();
}
void set_forward_header_key(const std::string& key) { fwd_header_key = key; }
typedef std::map<connection_hdl, websocket_connection_ptr,std::owner_less<connection_hdl> > con_map;
con_map _connections;
@ -313,14 +306,15 @@ namespace fc { namespace http {
on_connection_handler _on_connection;
fc::promise<void>::ptr _closed;
uint32_t _pending_messages = 0;
std::string fwd_header_key;
std::string _forward_header_key; // A header like "X-Forwarded-For" (XFF), with data IP:port
};
class websocket_tls_server_impl
{
public:
websocket_tls_server_impl( const string& server_pem, const string& ssl_password )
:_server_thread( fc::thread::current() )
websocket_tls_server_impl( const string& server_pem, const string& ssl_password,
const std::string& forward_header_key )
:_server_thread( fc::thread::current() ), _forward_header_key(forward_header_key)
{
{
_server.set_tls_init_handler( [=]( websocketpp::connection_hdl hdl ) -> context_ptr {
@ -345,7 +339,9 @@ namespace fc { namespace http {
_server.set_reuse_addr(true);
_server.set_open_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
auto new_con = std::make_shared<websocket_connection_impl<websocket_tls_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl) );
auto new_con = std::make_shared<websocket_connection_impl<
websocket_tls_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl),
_forward_header_key );
_on_connection( _connections[hdl] = new_con );
}).wait();
});
@ -355,8 +351,8 @@ namespace fc { namespace http {
assert( current_con != _connections.end() );
auto received = msg->get_payload();
std::shared_ptr<websocket_connection> con = current_con->second;
wlog( "Websocket TLS Server Remote: ${host} Payload: ${body}",
("host", con->get_remote_ip(fwd_header_key)) ("body", msg->get_payload()));
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",con->get_remote_endpoint()) ("msg",received) );
fc::async([con,received](){ con->on_message( received ); });
}).wait();
});
@ -364,14 +360,19 @@ namespace fc { namespace http {
_server.set_http_handler( [&]( connection_hdl hdl ){
_server_thread.async( [&](){
auto current_con = std::make_shared<websocket_connection_impl<websocket_tls_server_type::connection_ptr>>( _server.get_con_from_hdl(hdl) );
auto con = _server.get_con_from_hdl(hdl);
auto current_con = std::make_shared<websocket_connection_impl<
websocket_tls_server_type::connection_ptr>>( con, _forward_header_key );
try{
_on_connection( current_con );
auto con = _server.get_con_from_hdl(hdl);
wdump(("server")(con->get_request_body()));
std::string remote_endpoint = current_con->get_remote_endpoint();
std::string request_body = con->get_request_body();
wlog( "[IN] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",request_body) );
auto response = current_con->on_http( con->get_request_body() );
idump((response));
ilog( "[OUT] ${remote_endpoint} ${msg}",
("remote_endpoint",remote_endpoint) ("msg",response) );
con->set_body( std::move( response.body_as_string ) );
con->set_status( websocketpp::http::status_code::value( response.status ) );
} catch ( const fc::exception& e )
@ -413,8 +414,6 @@ namespace fc { namespace http {
_server.close( item.first, 0, "server exit" );
}
void set_forward_header_key( const std::string& key ) { fwd_header_key = key; }
typedef std::map<connection_hdl, websocket_connection_ptr,std::owner_less<connection_hdl> > con_map;
con_map _connections;
@ -422,7 +421,7 @@ namespace fc { namespace http {
websocket_tls_server_type _server;
on_connection_handler _on_connection;
fc::promise<void>::ptr _closed;
std::string fwd_header_key;
std::string _forward_header_key; // A header like "X-Forwarded-For" (XFF), with data IP:port
};
typedef websocketpp::client<asio_with_stub_log> websocket_client_type;
@ -558,8 +557,8 @@ namespace fc { namespace http {
} // namespace detail
websocket_server::websocket_server()
:my( new detail::websocket_server_impl() ) {}
websocket_server::websocket_server( const std::string& forward_header_key )
:my( new detail::websocket_server_impl( forward_header_key ) ) {}
websocket_server::~websocket_server(){}
void websocket_server::on_connection( const on_connection_handler& handler )
@ -597,12 +596,11 @@ namespace fc { namespace http {
my->_server.close(connection.first, websocketpp::close::status::normal, "Goodbye");
}
void websocket_server::set_forward_header_key( const std::string& key )
{
my->set_forward_header_key( key );
}
websocket_tls_server::websocket_tls_server( const string& server_pem, const string& ssl_password,
const std::string& forward_header_key )
:my( new detail::websocket_tls_server_impl(server_pem, ssl_password, forward_header_key) )
{}
websocket_tls_server::websocket_tls_server( const string& server_pem, const string& ssl_password ):my( new detail::websocket_tls_server_impl(server_pem, ssl_password) ) {}
websocket_tls_server::~websocket_tls_server(){}
void websocket_tls_server::on_connection( const on_connection_handler& handler )
@ -629,7 +627,11 @@ namespace fc { namespace http {
websocket_client::websocket_client( const std::string& ca_filename ):my( new detail::websocket_client_impl() ),smy(new detail::websocket_tls_client_impl( ca_filename )) {}
websocket_client::websocket_client( const std::string& ca_filename )
:my( new detail::websocket_client_impl() ),
smy(new detail::websocket_tls_client_impl( ca_filename ))
{}
websocket_client::~websocket_client(){ }
websocket_connection_ptr websocket_client::connect( const std::string& uri )
@ -646,7 +648,8 @@ namespace fc { namespace http {
my->_client.set_open_handler( [=]( websocketpp::connection_hdl hdl ){
my->_hdl = hdl;
auto con = my->_client.get_con_from_hdl(hdl);
my->_connection = std::make_shared<detail::websocket_connection_impl<detail::websocket_client_connection_type>>( con );
my->_connection = std::make_shared<detail::websocket_connection_impl<
detail::websocket_client_connection_type>>( con, "" );
my->_closed = promise<void>::create("websocket::closed");
my->_connected->set_value();
});
@ -654,7 +657,7 @@ namespace fc { namespace http {
auto con = my->_client.get_connection( uri, ec );
std::for_each(headers.begin(), headers.end(), [con](std::pair<std::string, std::string> in) {
con->append_header(in.first, in.second);
con->append_header(in.first, in.second);
});
if( ec ) FC_ASSERT( !ec, "error: ${e}", ("e",ec.message()) );
@ -675,7 +678,8 @@ namespace fc { namespace http {
smy->_client.set_open_handler( [=]( websocketpp::connection_hdl hdl ){
auto con = smy->_client.get_con_from_hdl(hdl);
smy->_connection = std::make_shared<detail::websocket_connection_impl<detail::websocket_tls_client_connection_type>>( con );
smy->_connection = std::make_shared<detail::websocket_connection_impl<
detail::websocket_tls_client_connection_type>>( con, "" );
smy->_closed = promise<void>::create("websocket::closed");
smy->_connected->set_value();
});
@ -708,14 +712,14 @@ namespace fc { namespace http {
websocket_connection_ptr websocket_tls_client::connect( const std::string& uri )
{ try {
// wlog( "connecting to ${uri}", ("uri",uri));
websocketpp::lib::error_code ec;
my->_connected = promise<void>::create("websocket::connect");
my->_client.set_open_handler( [=]( websocketpp::connection_hdl hdl ){
auto con = my->_client.get_con_from_hdl(hdl);
my->_connection = std::make_shared<detail::websocket_connection_impl<detail::websocket_tls_client_connection_type>>( con );
my->_connection = std::make_shared<detail::websocket_connection_impl<
detail::websocket_tls_client_connection_type>>( con, "" );
my->_closed = promise<void>::create("websocket::closed");
my->_connected->set_value();
});