|
@@ -171,6 +171,62 @@ void NodeIO::recv_commands(
|
|
});
|
|
});
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+/*
|
|
|
|
+ Handler for received client messages.
|
|
|
|
+
|
|
|
|
+*/
|
|
|
|
+void NetIO::handle_async_clients(std::shared_ptr<tcp::socket> csocket,
|
|
|
|
+ const boost::system::error_code& error, size_t auth_size,
|
|
|
|
+ size_t msgbundle_size)
|
|
|
|
+{
|
|
|
|
+ if(!error) {
|
|
|
|
+#ifdef VERBOSE_NET
|
|
|
|
+ printf("Accept handler success\n");
|
|
|
|
+#endif
|
|
|
|
+ // Read header (1 uint64_t) from the socket and extract the client ID
|
|
|
|
+ size_t header;
|
|
|
|
+ clientid_t cid;
|
|
|
|
+ boost::asio::read(*csocket,
|
|
|
|
+ boost::asio::buffer(&header, sizeof(uint64_t)));
|
|
|
|
+
|
|
|
|
+ if((header & 0xff) == CLIENT_AUTHENTICATE) {
|
|
|
|
+ // Read the authentication token
|
|
|
|
+ boost::asio::read(*csocket,
|
|
|
|
+ boost::asio::buffer(&header, auth_size));
|
|
|
|
+
|
|
|
|
+ } else if ((header & 0xff) == CLIENT_MESSAGE_BUNDLE) {
|
|
|
|
+ unsigned char *msgbundle = (unsigned char*) malloc(msgbundle_size);
|
|
|
|
+ cid = (clientid_t)(header >> 8);
|
|
|
|
+
|
|
|
|
+ // Read the message_bundle
|
|
|
|
+ boost::asio::read(*csocket,
|
|
|
|
+ boost::asio::buffer(msgbundle, msgbundle_size));
|
|
|
|
+
|
|
|
|
+ //Ingest the message_bundle
|
|
|
|
+ bool ret = ecall_ingest_msgbundle(cid, msgbundle, apiparams.m_priv_out);
|
|
|
|
+ free(msgbundle);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+ start_accept(auth_size, msgbundle_size);
|
|
|
|
+ } else {
|
|
|
|
+ printf("Accept handler failed\n");
|
|
|
|
+ }
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
+/*
|
|
|
|
+ Asynchronously accept client connections
|
|
|
|
+*/
|
|
|
|
+void NetIO::start_accept(size_t auth_size, size_t msgbundle_size)
|
|
|
|
+{
|
|
|
|
+ std::shared_ptr<tcp::socket> csocket(new tcp::socket(io_context()));
|
|
|
|
+#ifdef VERBOSE_NET
|
|
|
|
+ std::cout << "Accepting on " << myconf.clistenhost << ":" << myconf.clistenport << "\n";
|
|
|
|
+#endif
|
|
|
|
+ client_acceptor->async_accept(*csocket,
|
|
|
|
+ boost::bind(&NetIO::handle_async_clients, this, csocket,
|
|
|
|
+ boost::asio::placeholders::error, auth_size, msgbundle_size));
|
|
|
|
+}
|
|
|
|
+
|
|
|
|
|
|
NetIO::NetIO(boost::asio::io_context &io_context, const Config &config)
|
|
NetIO::NetIO(boost::asio::io_context &io_context, const Config &config)
|
|
: context(io_context), conf(config),
|
|
: context(io_context), conf(config),
|
|
@@ -241,8 +297,24 @@ NetIO::NetIO(boost::asio::io_context &io_context, const Config &config)
|
|
std::cerr << "Connected to " << config.nodes[i].name << "\n";
|
|
std::cerr << "Connected to " << config.nodes[i].name << "\n";
|
|
#endif
|
|
#endif
|
|
}
|
|
}
|
|
|
|
+
|
|
|
|
+ if(myconf.roles & ROLE_INGESTION) {
|
|
|
|
+ client_acceptor = std::shared_ptr<tcp::acceptor>(
|
|
|
|
+ new tcp::acceptor(io_context,
|
|
|
|
+ resolver.resolve(this->myconf.clistenhost,
|
|
|
|
+ this->myconf.clistenport)->endpoint()));
|
|
|
|
+
|
|
|
|
+ size_t auth_size, msgbundle_size;
|
|
|
|
+ auth_size = SGX_AESGCM_MAC_SIZE;
|
|
|
|
+ msgbundle_size = SGX_AESGCM_IV_SIZE +
|
|
|
|
+ (apiparams.m_priv_out * apiparams.msg_size) + SGX_AESGCM_MAC_SIZE;
|
|
|
|
+ start_accept(auth_size, msgbundle_size);
|
|
|
|
+ }
|
|
|
|
+
|
|
|
|
+
|
|
}
|
|
}
|
|
|
|
|
|
|
|
+
|
|
void NetIO::recv_commands(
|
|
void NetIO::recv_commands(
|
|
std::function<void(boost::system::error_code)> error_cb,
|
|
std::function<void(boost::system::error_code)> error_cb,
|
|
std::function<void(uint32_t)> epoch_cb)
|
|
std::function<void(uint32_t)> epoch_cb)
|