Merge branch 'develop' into readme-updates

ExitRequestedException.cpp

+/*
+    Copyright (C) 2019-Present SKALE Labs
+
+    This file is part of sgxwallet.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file ExitRequestedException.cpp
+    @author Stan Kladko
+    @date 2018
+*/
+
+#include "ExitRequestedException.h"
+
+
+ExitRequestedException::ExitRequestedException() {
+
+}

ExitRequestedException.h

+/*
+    Copyright (C) 2019-Present SKALE Labs
+
+    This file is part of sgxwallet.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file ExitRequestedException.h
+    @author Stan Kladko
+    @date 2018
+*/
+
+#pragma once
+
+#include <exception>
+
+class ExitRequestedException : public std::exception {
+public:
+    ExitRequestedException();
+};

Makefile.am

@@ -71,9 +71,9 @@ bin_PROGRAMS = sgxwallet testw sgx_util
 
 
 COMMON_SRC = SGXException.cpp ExitHandler.cpp zmq_src/ZMQClient.cpp zmq_src/RspMessage.cpp zmq_src/ReqMessage.cpp \
-             zmq_src/ZMQMessage.cpp zmq_src/ZMQServer.cpp \
+             zmq_src/ZMQMessage.cpp zmq_src/ZMQServer.cpp zmq_src/Agent.cpp  zmq_src/WorkerThreadPool.cpp ExitRequestedException.cpp \
              InvalidStateException.cpp Exception.cpp InvalidArgumentException.cpp Log.cpp TECrypto.cpp \
-             SGXWalletServer.cpp SGXRegistrationServer.cpp CSRManagerServer.cpp BLSCrypto.cpp CryptoTools.cpp \
+             SGXWalletServer.cpp  SGXRegistrationServer.cpp CSRManagerServer.cpp BLSCrypto.cpp CryptoTools.cpp \
              DKGCrypto.cpp ServerInit.cpp BLSPrivateKeyShareSGX.cpp LevelDB.cpp ServerDataChecker.cpp SEKManager.cpp \
              third_party/intel/sgx_stub.c third_party/intel/sgx_detect_linux.c third_party/intel/create_enclave.c \
              third_party/intel/oc_alloc.c ECDSAImpl.c TestUtils.cpp sgxwallet.c SGXInfoServer.cpp ECDSACrypto.cpp

VERSION

-1.78.0
\ No newline at end of file
+1.83.0
\ No newline at end of file

common.h

@@ -31,16 +31,20 @@ using namespace std;
 #include <iostream>
 #include <map>
 #include <memory>
-
 #include <sys/types.h>
 #include <sys/sysinfo.h>
 #include <string.h>
-
 #include <vector>
-
+#include <json/value.h>
 #include <boost/throw_exception.hpp>
-
 #include <gmp.h>
+#include <thread>
+#include <functional>
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+
+
 #include "secure_enclave/Verify.h"
 #include "InvalidStateException.h"
 #include "SGXException.h"

secure_enclave/AESUtils.c

@@ -49,11 +49,16 @@ int AES_encrypt(char *message, uint8_t *encr_message, uint64_t encrBufLen, unsig
         return -2;
     }
 
+    if (!resultLen) {
+        LOG_ERROR("Null resultLen in AES_encrypt");
+        return -3;
+    }
+
     uint64_t len = strlen(message) + 1;
 
     if (2 + len + SGX_AESGCM_MAC_SIZE + SGX_AESGCM_IV_SIZE > encrBufLen ) {
         LOG_ERROR("Output buffer too small");
-        return -3;
+        return -4;
     }
 
     SAFE_CHAR_BUF(fullMessage, len + 2);
@@ -97,36 +102,36 @@ int AES_decrypt(uint8_t *encr_message, uint64_t length, char *message, uint64_t
         return -3;
     }
 
-    if (!encr_message) {
+    if (!exportable) {
         LOG_ERROR("Null exportable in AES_encrypt");
         return -4;
     }
 
 
-  if (length < SGX_AESGCM_MAC_SIZE + SGX_AESGCM_IV_SIZE) {
-      LOG_ERROR("length < SGX_AESGCM_MAC_SIZE - SGX_AESGCM_IV_SIZE");
-      return -1;
-  }
+    if (length < SGX_AESGCM_MAC_SIZE + SGX_AESGCM_IV_SIZE) {
+        LOG_ERROR("length < SGX_AESGCM_MAC_SIZE - SGX_AESGCM_IV_SIZE");
+        return -5;
+    }
 
-  uint64_t len = length - SGX_AESGCM_MAC_SIZE - SGX_AESGCM_IV_SIZE;
+    uint64_t len = length - SGX_AESGCM_MAC_SIZE - SGX_AESGCM_IV_SIZE;
 
-  if (msgLen < len) {
+    if (msgLen < len) {
         LOG_ERROR("Output buffer not large enough");
-        return -2;
-  }
-
-  sgx_status_t status = sgx_rijndael128GCM_decrypt(&(AES_key[512]),
-                                                   encr_message + SGX_AESGCM_MAC_SIZE + SGX_AESGCM_IV_SIZE, len,
-                                                   (unsigned char*) message,
-                                                   encr_message + SGX_AESGCM_MAC_SIZE, SGX_AESGCM_IV_SIZE,
-                                                   NULL, 0,
-                                                   (sgx_aes_gcm_128bit_tag_t *)encr_message);
-
-  *type = message[0];
-  *exportable = message[1];
-  for (int i = 2; i < strlen(message) + 1; i++) {
-      message[i - 2 ] = message[i];
-  }
-
-  return status;
+        return -6;
+    }
+
+    sgx_status_t status = sgx_rijndael128GCM_decrypt(&(AES_key[512]),
+                                                    encr_message + SGX_AESGCM_MAC_SIZE + SGX_AESGCM_IV_SIZE, len,
+                                                    (unsigned char*) message,
+                                                    encr_message + SGX_AESGCM_MAC_SIZE, SGX_AESGCM_IV_SIZE,
+                                                    NULL, 0,
+                                                    (sgx_aes_gcm_128bit_tag_t *)encr_message);
+
+    *type = message[0];
+    *exportable = message[1];
+    for (int i = 2; i < strlen(message) + 1; i++) {
+        message[i - 2 ] = message[i];
+    }
+
+    return status;
 }

secure_enclave/secure_enclave.c

@@ -597,7 +597,6 @@ void trustedEcdsaSign(int *errStatus, char *errString, uint8_t *encryptedPrivate
     LOG_DEBUG("SGX call completed");
 }
 
-
 void trustedDecryptKey(int *errStatus, char *errString, uint8_t *encryptedPrivateKey,
                           uint64_t enc_len, char *key) {
 
@@ -606,24 +605,14 @@ void trustedDecryptKey(int *errStatus, char *errString, uint8_t *encryptedPrivat
 
     CHECK_STATE(encryptedPrivateKey);
     CHECK_STATE(key);
+    CHECK_STATE( enc_len == strnlen( encryptedPrivateKey, 1024 ) );
 
     *errStatus = -9;
 
     uint8_t type = 0;
     uint8_t exportable = 0;
 
-    int status = AES_decrypt(encryptedPrivateKey, enc_len, key, 3072,
-                             &type, &exportable);
-
-    if (exportable != EXPORTABLE) {
-        while (*key != '\0') {
-            *key++ = '0';
-        }
-        *errStatus = -11;
-        snprintf(errString, BUF_LEN, "Key is not exportable");
-        LOG_ERROR(errString);
-        goto clean;
-    }
+    int status = AES_decrypt(encryptedPrivateKey, enc_len, key, 1024, &type, &exportable);
 
     if (status != 0) {
         *errStatus = status;
@@ -632,22 +621,30 @@ void trustedDecryptKey(int *errStatus, char *errString, uint8_t *encryptedPrivat
         goto clean;
     }
 
-    *errStatus = -10;
-
-    uint64_t keyLen = strnlen(key, MAX_KEY_LENGTH);
+    size_t keyLen = strnlen(key, MAX_KEY_LENGTH);
 
     if (keyLen == MAX_KEY_LENGTH) {
+        *errStatus = -10;
         snprintf(errString, BUF_LEN, "Key is not null terminated");
         LOG_ERROR(errString);
         goto clean;
     }
 
+    if (exportable != EXPORTABLE) {
+        while (*key != '\0') {
+            *key++ = '0';
+        }
+        *errStatus = -11;
+        snprintf(errString, BUF_LEN, "Key is not exportable");
+        LOG_ERROR(errString);
+        goto clean;
+    }
+
     SET_SUCCESS
     clean:
     ;
 }
 
-
 void trustedEncryptKey(int *errStatus, char *errString, const char *key,
                           uint8_t *encryptedPrivateKey, uint64_t *enc_len) {
     LOG_INFO(__FUNCTION__);

zmq_src/Agent.cpp

+/*
+    Copyright (C) 2021 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file Agent.cpp
+    @author Stan Kladko
+    @date 2021
+*/
+
+#include "common.h"
+#include "Agent.h"
+
+Agent::Agent() : startedWorkers(false) {};
+
+void Agent::waitOnGlobalStartBarrier() {
+    unique_lock<mutex> lock(startMutex);
+    while (!startedWorkers) {
+        // wait until notified to start running
+        startCond.wait(lock);
+    }
+}
+
+Agent::~Agent() {
+}
+
+
+void Agent::releaseWorkers() {
+
+    if (startedWorkers.exchange(true)) {
+        // already started
+        return;
+    }
+
+    lock_guard<mutex> lock(startMutex);
+
+    startCond.notify_all();
+}

zmq_src/Agent.h

+/*
+    Copyright (C) 2021 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file Agent.h
+    @author Stan Kladko
+    @date 2021
+*/
+
+#pragma once
+
+class Agent {
+
+protected:
+
+    atomic_bool startedWorkers;
+
+    mutex messageMutex;
+    condition_variable messageCond;
+
+    condition_variable startCond;
+    mutex startMutex;
+
+    recursive_mutex m;
+
+
+public:
+
+    Agent();
+
+    virtual ~Agent();
+
+    void releaseWorkers();
+
+    void waitOnGlobalStartBarrier();
+
+    recursive_mutex& getMainMutex() { return m; }
+};

zmq_src/WorkerThreadPool.cpp

+/*
+    Copyright (C) 2021 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file WorkerThreadPool.cpp
+    @author Stan Kladko
+    @date 2021
+*/
+
+#include "document.h"
+#include "stringbuffer.h"
+#include "writer.h"
+
+
+
+#include "common.h"
+#include "sgxwallet_common.h"
+#include "third_party/spdlog/spdlog.h"
+#include "ZMQServer.h"
+#include "WorkerThreadPool.h"
+
+
+WorkerThreadPool::WorkerThreadPool(uint64_t _numThreads, ZMQServer *_agent) : joined(false) {
+    CHECK_STATE(_numThreads > 0);
+    CHECK_STATE(_agent);
+
+    spdlog::info("Creating thread pool. Threads count:" + to_string(_numThreads));
+
+    this->agent = _agent;
+    this->numThreads = _numThreads;;
+
+
+    for (uint64_t i = 0; i < (uint64_t) numThreads; i++) {
+        createThread(i);
+    }
+
+    spdlog::info("Created thread pool");
+
+}
+
+
+void WorkerThreadPool::joinAll() {
+
+    spdlog::info("Joining worker threads ...");
+
+    if (joined.exchange(true))
+        return;
+
+    for (auto &&thread : threadpool) {
+        if (thread->joinable())
+            thread->join();
+        CHECK_STATE(!thread->joinable());
+    }
+
+    spdlog::info("Joined worker threads.");
+}
+
+bool WorkerThreadPool::isJoined() const {
+    return joined;
+}
+
+WorkerThreadPool::~WorkerThreadPool(){
+}
+
+void WorkerThreadPool::createThread(uint64_t _threadNumber) {
+
+    spdlog::info("Starting ZMQ worker thread " + to_string(_threadNumber) );
+
+    this->threadpool.push_back(
+            make_shared< thread >( ZMQServer::workerThreadMessageProcessLoop, agent, _threadNumber ) );
+
+    spdlog::info("Started ZMQ worker thread " + to_string(_threadNumber) );
+}

zmq_src/WorkerThreadPool.h

+/*
+    Copyright (C) 2018-2019 SKALE Labs
+
+    This file is part of skale-consensus.
+
+    skale-consensus is free software: you can redistribute it and/or modify
+    it under the terms of the GNU Affero General Public License as published
+    by the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    skale-consensus is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU Affero General Public License for more details.
+
+    You should have received a copy of the GNU Affero General Public License
+    along with skale-consensus.  If not, see <https://www.gnu.org/licenses/>.
+
+    @file WorkerThreadPool.h
+    @author Stan Kladko
+    @date 2018
+*/
+
+#pragma once
+
+class Agent;
+class ZMQServer;
+
+class WorkerThreadPool {
+
+    void createThread( uint64_t threadNumber );
+
+    recursive_mutex m;
+
+protected:
+
+    atomic_bool joined;
+    vector<shared_ptr<thread>> threadpool;
+
+    uint64_t numThreads = 0;
+    ZMQServer* agent = nullptr;
+
+public:
+
+    WorkerThreadPool(uint64_t _numThreads, ZMQServer *_agent);
+
+    virtual ~WorkerThreadPool();
+
+    void joinAll();
+
+    bool isJoined() const;
+
+};

zmq_src/ZMQClient.h

@@ -31,7 +31,6 @@
 #include <openssl/rand.h>
 
 #include "third_party/spdlog/spdlog.h"
-
 #include <zmq.hpp>
 #include "zhelpers.hpp"
 #include <jsonrpccpp/client.h>

zmq_src/ZMQMessage.h

@@ -24,9 +24,6 @@
 #pragma once
 
 
-#include <memory>
-#include <vector>
-
 #include <openssl/pem.h>
 #include <openssl/evp.h>
 #include <openssl/err.h>

zmq_src/ZMQServer.h

@@ -26,30 +26,50 @@
 #define SGXWALLET_ZMQServer_H
 
 
-#include <vector>
-#include <thread>
-#include <memory>
-#include <functional>
-#include <atomic>
+#include "third_party/readerwriterqueue.h"
+#include "third_party/concurrentqueue.h"
+
 
 #include <zmq.hpp>
 #include "zhelpers.hpp"
 
-using namespace std;
+#include "Agent.h"
+#include "WorkerThreadPool.h"
+#include "ZMQMessage.h"
+
+using namespace moodycamel;
+
+static const uint64_t NUM_ZMQ_WORKER_THREADS = 16;
 
 
-class ZMQServer {
+class ZMQServer : public Agent{
 
     uint64_t workerThreads;
 
+    string caCertFile;
+    string caCert;
+
+    ConcurrentQueue<pair<Json::Value, shared_ptr<zmq::message_t>>> outgoingQueue;
+
+    vector<BlockingReaderWriterQueue<pair<shared_ptr<ZMQMessage>, shared_ptr<zmq::message_t>>>> incomingQueue;
+
+    bool checkKeyOwnership = true;
+
+    shared_ptr<zmq::context_t> ctx;
+    shared_ptr<zmq::socket_t> socket;
+
+    static atomic<bool> isExitRequested;
+
+    void doOneServerLoop();
+
 public:
 
     bool checkSignature = false;
-    string caCertFile = "";
-    string caCert = "";
 
     static shared_ptr<ZMQServer> zmqServer;
 
+    shared_ptr<WorkerThreadPool> threadPool = nullptr;
+
     static shared_ptr<std::thread> serverThread;
 
     ZMQServer(bool _checkSignature, bool _checkKeyOwnership, const string& _caCertFile);
@@ -58,18 +78,24 @@ public:
 
     void run();
 
+    void initListenSocket();
+
     static void initZMQServer(bool _checkSignature, bool _checkKeyOwnership);
     static void exitZMQServer();
 
-private:
-    bool checkKeyOwnership = true;
+    static void workerThreadMessageProcessLoop(ZMQServer* agent, uint64_t _threadNumber );
 
-    shared_ptr<zmq::context_t> ctx;
-    shared_ptr<zmq::socket_t> socket;
+    void workerThreadProcessNextMessage(uint64_t _threadNumber);
 
-    static std::atomic<bool> isExitRequested;
+    void checkForExit();
 
-    void doOneServerLoop();
+    void waitForIncomingAndProcessOutgoingMessages();
+
+    pair<string, shared_ptr<zmq::message_t>>  receiveMessage();
+
+    void sendToClient(Json::Value& _result,  shared_ptr<zmq::message_t>& _identity);
+
+    void sendMessagesInOutgoingMessageQueueIfAny();
 
 };