Allow tx without budget

bftengine/include/bftengine/Replica.hpp

@@ -28,7 +28,7 @@
 #include "PersistentStorage.hpp"
 #include "IRequestHandler.hpp"
 #include "InternalBFTClient.hpp"
-
+#include "Timers.hpp"
 namespace concord::cron {
 class TicksGenerator;
 }
@@ -38,7 +38,10 @@ class ISecretsManagerImpl;
 }
 
 namespace bftEngine {
-
+namespace impl {
+class MsgsCommunicator;
+class MsgHandlersRegistrator;
+}  // namespace impl
 // Possible values for 'flags' parameter
 enum MsgFlag : uint64_t {
   EMPTY_FLAGS = 0x0,
@@ -48,6 +51,7 @@ enum MsgFlag : uint64_t {
   KEY_EXCHANGE_FLAG = 0x8,  // TODO [TK] use reconfig_flag
   TICK_FLAG = 0x10,
   RECONFIG_FLAG = 0x20,
+  INTERNAL_FLAG = 0x40,
   PUBLISH_ON_CHAIN_OBJECT_FLAG = 0x80,
   CLIENTS_PUB_KEYS_FLAG = 0x100,
   DB_CHECKPOINT_FLAG = 0x200
@@ -135,6 +139,10 @@ class IReplica {
 
   // Returns the internal persistent storage object.
   virtual std::shared_ptr<impl::PersistentStorage> persistentStorage() const = 0;
+
+  virtual std::shared_ptr<impl::MsgsCommunicator> getMsgsCommunicator() const { return nullptr; }
+  virtual std::shared_ptr<impl::MsgHandlersRegistrator> getMsgHandlersRegistrator() const { return nullptr; }
+  virtual concordUtil::Timers *getTimers() { return nullptr; }
 };
 
 }  // namespace bftEngine

bftengine/src/bftengine/BFTEngine.cpp

@@ -69,6 +69,12 @@ class ReplicaInternal : public IReplica {
 
   std::shared_ptr<IInternalBFTClient> internalClient() const override { return internal_client_; }
 
+  std::shared_ptr<MsgsCommunicator> getMsgsCommunicator() const override { return replica_->getMsgsCommunicator(); }
+  std::shared_ptr<MsgHandlersRegistrator> getMsgHandlersRegistrator() const override {
+    return replica_->getMsgHandlersRegistrator();
+  }
+  concordUtil::Timers *getTimers() override { return replica_->getTimers(); }
+
  private:
   std::unique_ptr<ReplicaBase> replica_;
   std::condition_variable debugWait_;

bftengine/src/bftengine/ReplicaBase.hpp

@@ -52,7 +52,7 @@ class ReplicaBase {
 
   std::shared_ptr<MsgsCommunicator> getMsgsCommunicator() const { return msgsCommunicator_; }
   std::shared_ptr<MsgHandlersRegistrator> getMsgHandlersRegistrator() const { return msgHandlers_; }
-
+  concordUtil::Timers* getTimers() { return &timers_; }
   void SetAggregator(std::shared_ptr<concordMetrics::Aggregator> aggregator) {
     if (aggregator) {
       aggregator_ = aggregator;

bftengine/src/bftengine/ReplicaImp.cpp

@@ -438,10 +438,10 @@ void ReplicaImp::onMessage<ClientRequestMsg>(ClientRequestMsg *m) {
 
   // check message validity
   const bool invalidClient =
-      !isValidClient(clientId) &&
-      !((repsInfo->isIdOfReplica(clientId) || repsInfo->isIdOfPeerRoReplica(clientId)) && (flags & RECONFIG_FLAG));
+      !isValidClient(clientId) && !((repsInfo->isIdOfReplica(clientId) || repsInfo->isIdOfPeerRoReplica(clientId)) &&
+                                    (flags & RECONFIG_FLAG || flags & INTERNAL_FLAG));
   const bool sentFromReplicaToNonPrimary =
-      !(flags & RECONFIG_FLAG) && repsInfo->isIdOfReplica(senderId) && !isCurrentPrimary();
+      !(flags & RECONFIG_FLAG || flags & INTERNAL_FLAG) && repsInfo->isIdOfReplica(senderId) && !isCurrentPrimary();
 
   if (invalidClient) {
     ++numInvalidClients;
@@ -5487,7 +5487,9 @@ void ReplicaImp::executeRequestsInPrePrepareMsg(concordUtils::SpanWrapper &paren
           reqIdx++;
           continue;
         }
-        const bool validClient = isValidClient(clientId) || ((req.flags() & RECONFIG_FLAG) && isIdOfReplica(clientId));
+        const bool validClient =
+            isValidClient(clientId) ||
+            ((req.flags() & RECONFIG_FLAG || req.flags() & INTERNAL_FLAG) && isIdOfReplica(clientId));
         if (!validClient) {
           ++numInvalidClients;
           LOG_WARN(CNSUS, "The client is not valid" << KVLOG(clientId));

bftengine/src/bftengine/messages/ClientRequestMsg.cpp

@@ -100,7 +100,8 @@ bool ClientRequestMsg::shouldValidateAsync() const {
   // manner, as that will lead to overhead. Similarly, key exchanges should happen rarely, and thus we should validate
   // as quick as possible, in sync.
   const auto* header = msgBody();
-  if (((header->flags & RECONFIG_FLAG) != 0) || ((header->flags & KEY_EXCHANGE_FLAG) != 0)) {
+  if (((header->flags & RECONFIG_FLAG) != 0) || ((header->flags & KEY_EXCHANGE_FLAG) != 0) ||
+      (header->flags & INTERNAL_FLAG) != 0) {
     return false;
   }
   return true;
@@ -119,7 +120,8 @@ void ClientRequestMsg::validateImp(const ReplicasInfo& repInfo) const {
   }
 
   PrincipalId clientId = header->idOfClientProxy;
-  if ((header->flags & RECONFIG_FLAG) == 0) ConcordAssert(this->senderId() != repInfo.myId());
+  if ((header->flags & RECONFIG_FLAG) == 0 && (header->flags & INTERNAL_FLAG) == 0)
+    ConcordAssert(this->senderId() != repInfo.myId());
 
   /// to do - should it be just the header?
   auto minMsgSize = sizeof(ClientRequestMsgHeader) + header->cidLength + spanContextSize() + header->reqSignatureLength;
@@ -135,9 +137,9 @@ void ClientRequestMsg::validateImp(const ReplicasInfo& repInfo) const {
   bool isIdOfExternalClient = repInfo.isIdOfExternalClient(clientId);
   bool doSigVerify = false;
   bool emptyReq = (header->requestLength == 0);
-  if ((header->flags & RECONFIG_FLAG) != 0 &&
+  if (((header->flags & RECONFIG_FLAG) != 0 || (header->flags & INTERNAL_FLAG) != 0) &&
       (repInfo.isIdOfReplica(clientId) || repInfo.isIdOfPeerRoReplica(clientId))) {
-    // Allow every reconfiguration message from replicas (it will be verified in the reconfiguration handler)
+    // Allow every reconfiguration/internal message from replicas (it will be verified in the reconfiguration handler)
     return;
   }
   if (!repInfo.isValidPrincipalId(clientId)) {

kvbc/include/Replica.h

@@ -181,6 +181,11 @@ class Replica : public IReplica,
   }
 
   virtual ~Replica();
+  std::shared_ptr<MsgsCommunicator> getMsgsCommunicator() const { return m_replicaPtr->getMsgsCommunicator(); }
+  std::shared_ptr<MsgHandlersRegistrator> getMsgHandlersRegistrator() const {
+    return m_replicaPtr->getMsgHandlersRegistrator();
+  }
+  concordUtil::Timers *getTimers() { return m_replicaPtr->getTimers(); }
 
  protected:
   RawBlock getBlockInternal(BlockId blockId) const;

utt-replica/signature-processor/include/sigProcessor.hpp

@@ -36,6 +36,26 @@ class SigProcessor {
    */
   using GenerateAppClientRequestCb = std::function<std::vector<uint8_t>(uint64_t, const std::vector<uint8_t>&)>;
 
+  class CompleteSignatureMsg {
+    std::map<uint32_t, std::vector<uint8_t>> sigs;
+    std::vector<uint8_t> full_sig;
+    uint32_t num_replicas{0};
+
+   public:
+    CompleteSignatureMsg(uint32_t n, const std::map<uint32_t, std::vector<uint8_t>>&, const std::vector<uint8_t>&);
+    explicit CompleteSignatureMsg() = default;
+    CompleteSignatureMsg(const std::vector<uint8_t>& buffer);
+    bool validate() const;
+    const std::map<uint32_t, std::vector<uint8_t>>& getPartialSigs() const;
+    const std::vector<uint8_t>& getFullSig() const;
+    /*
+      The serialized output is in the following format:
+      [num_replicas|full_sig.size()|full_sig|sigs.size()|s.size()|s] (for s in sigs)
+    */
+    std::vector<uint8_t> serialize() const;
+    CompleteSignatureMsg& deserialize(const std::vector<uint8_t>&);
+  };
+
   static const GenerateAppClientRequestCb default_client_app_request_generator;
 
  private:
@@ -136,13 +156,9 @@ class SigProcessor {
   /**
    * @brief Publishing the complete signature to the consensus
    *
-   * @param sig_id the signature id
-   * @param complete_signature the complete aggregated signature
-   * @param cb a callback that defines how the upper level generates the application request to the consensus
+   * @param job_entry The relevant job
    */
-  void publishCompleteSignature(uint64_t sig_id,
-                                const std::vector<uint8_t>& complete_signature,
-                                const GenerateAppClientRequestCb& cb);
+  void publishCompleteSignature(const SigJobEntry& job_entry);
   /**
    * @brief Handles the event of timeout for a specific job
    *

utt-replica/signature-processor/src/sigProcessor.cpp

@@ -17,6 +17,7 @@
 #include "bftengine/ReplicaConfig.hpp"
 #include "bftengine/messages/MessageBase.hpp"
 #include "bftengine/messages/ClientRequestMsg.hpp"
+#include "bftengine/Replica.hpp"
 #include <algorithm>
 
 namespace utt {
@@ -141,8 +142,7 @@ void SigProcessor::processSignature(uint64_t sig_id,
 #endif
     // if we already have enough valid partial signatures, lets publish the complete signature and return
     if (entry->partial_sigs.size() == threshold_) {
-      auto complete_sig = libutt::api::Utils::aggregateSigShares(n_, entry->partial_sigs);
-      publishCompleteSignature(sig_id, complete_sig, entry->client_app_data_generator_cb_);
+      publishCompleteSignature(*entry);
       return;
     }
   }
@@ -179,26 +179,26 @@ void SigProcessor::onReceivingNewPartialSig(uint64_t sig_id,
     // We don't care doing this part under the entry lock, because if we did reach the threshold, we won't use this
     // entry anymore
     if (entry->client_app_data_generator_cb_ != nullptr && entry->partial_sigs.size() == threshold_) {
-      auto sig = libutt::api::Utils::aggregateSigShares(n_, entry->partial_sigs);
-      publishCompleteSignature(sig_id, sig, entry->client_app_data_generator_cb_);
+      publishCompleteSignature(*entry);
     }
   }
 }
-void SigProcessor::publishCompleteSignature(uint64_t sig_id,
-                                            const std::vector<uint8_t>& sig,
-                                            const GenerateAppClientRequestCb& cb) {
+
+void SigProcessor::publishCompleteSignature(const SigJobEntry& job_entry) {
+  auto sig_id = job_entry.job_id;
+  auto fsig = libutt::api::Utils::aggregateSigShares(n_, job_entry.partial_sigs);
+  CompleteSignatureMsg msg(n_, job_entry.partial_sigs, fsig);
   auto requestSeqNum =
       std::chrono::duration_cast<std::chrono::microseconds>(getMonotonicTime().time_since_epoch()).count();
-  std::vector<uint8_t> appClientReq = cb(sig_id, sig);
-  std::unique_ptr<MessageBase> cmsg =
-      std::make_unique<bftEngine::impl::ClientRequestMsg>(repId_,
-                                                          0x0,
-                                                          requestSeqNum,
-                                                          (uint32_t)appClientReq.size(),
-                                                          (const char*)appClientReq.data(),
-                                                          60000,
-                                                          "new-utt-sig-" + std::to_string(sig_id));
-  msgs_communicator_->getIncomingMsgsStorage()->pushExternalMsg(std::move(cmsg));
+  std::vector<uint8_t> appClientReq = job_entry.client_app_data_generator_cb_(sig_id, msg.serialize());
+  auto crm = std::make_unique<bftEngine::impl::ClientRequestMsg>(repId_,
+                                                                 bftEngine::MsgFlag::INTERNAL_FLAG,
+                                                                 requestSeqNum,
+                                                                 (uint32_t)appClientReq.size(),
+                                                                 (const char*)appClientReq.data(),
+                                                                 60000,
+                                                                 "new-utt-sig-" + std::to_string(sig_id));
+  msgs_communicator_->getIncomingMsgsStorage()->pushExternalMsg(std::move(crm));
 }
 // Called by the validating thread
 void SigProcessor::onReceivingNewValidFullSig(uint64_t sig_id) {
@@ -227,4 +227,73 @@ void SigProcessor::onJobTimeout(uint64_t job_id, const std::vector<uint8_t>& sig
     msgs_communicator_->sendAsyncMessage((uint64_t)rid, msg.body(), msg.size());
   }
 }
+
+SigProcessor::CompleteSignatureMsg::CompleteSignatureMsg(uint32_t n,
+                                                         const std::map<uint32_t, std::vector<uint8_t>>& psigs,
+                                                         const std::vector<uint8_t>& fsig)
+    : sigs{psigs}, full_sig{fsig}, num_replicas{n} {}
+bool SigProcessor::CompleteSignatureMsg::validate() const {
+  auto complete_sig = libutt::api::Utils::aggregateSigShares(num_replicas, sigs);
+  return full_sig == complete_sig;
+}
+SigProcessor::CompleteSignatureMsg::CompleteSignatureMsg(const std::vector<uint8_t>& buffer) { deserialize(buffer); }
+const std::vector<uint8_t>& SigProcessor::CompleteSignatureMsg::getFullSig() const { return full_sig; }
+const std::map<uint32_t, std::vector<uint8_t>>& SigProcessor::CompleteSignatureMsg::getPartialSigs() const {
+  return sigs;
+}
+std::vector<uint8_t> SigProcessor::CompleteSignatureMsg::serialize() const {
+  uint64_t loc{0};
+  uint64_t fsig_size = full_sig.size();
+  uint64_t psigs_size = sigs.size();
+  size_t msg_size = sizeof(num_replicas) + sizeof(fsig_size) + full_sig.size();
+  msg_size += sizeof(psigs_size);
+  for (const auto& [k, v] : sigs) msg_size += (sizeof(k) + sizeof(uint64_t) + v.size());
+  std::vector<uint8_t> ret(msg_size);
+  std::memcpy(ret.data(), &num_replicas, sizeof(num_replicas));
+  loc += sizeof(num_replicas);
+  std::memcpy(ret.data() + loc, &fsig_size, sizeof(fsig_size));
+  loc += sizeof(fsig_size);
+  std::memcpy(ret.data() + loc, full_sig.data(), full_sig.size());
+  loc += full_sig.size();
+  std::memcpy(ret.data() + loc, &psigs_size, sizeof(psigs_size));
+  loc += sizeof(psigs_size);
+  for (const auto& [k, v] : sigs) {
+    std::memcpy(ret.data() + loc, &k, sizeof(k));
+    loc += sizeof(k);
+    uint64_t ssize = v.size();
+    std::memcpy(ret.data() + loc, &ssize, sizeof(ssize));
+    loc += sizeof(ssize);
+    std::memcpy(ret.data() + loc, v.data(), v.size());
+    loc += v.size();
+  }
+  return ret;
+}
+
+SigProcessor::CompleteSignatureMsg& SigProcessor::CompleteSignatureMsg::deserialize(
+    const std::vector<uint8_t>& buffer) {
+  size_t loc{0};
+  std::memcpy(&num_replicas, buffer.data(), sizeof(num_replicas));
+  loc += sizeof(num_replicas);
+  uint64_t fsig_size{0};
+  std::memcpy(&fsig_size, buffer.data() + loc, sizeof(fsig_size));
+  loc += sizeof(fsig_size);
+  full_sig.resize(fsig_size);
+  std::memcpy(full_sig.data(), buffer.data() + loc, fsig_size);
+  loc += fsig_size;
+  uint64_t psigs_nums{0};
+  std::memcpy(&psigs_nums, buffer.data() + loc, sizeof(psigs_nums));
+  loc += sizeof(psigs_nums);
+  for (size_t i = 0; i < psigs_nums; i++) {
+    uint32_t rep_id{0};
+    std::memcpy(&rep_id, buffer.data() + loc, sizeof(rep_id));
+    loc += sizeof(rep_id);
+    uint64_t s_size{0};
+    std::memcpy(&s_size, buffer.data() + loc, sizeof(s_size));
+    loc += sizeof(s_size);
+    sigs[rep_id] = std::vector<uint8_t>(s_size);
+    std::memcpy(sigs[rep_id].data(), buffer.data() + loc, s_size);
+    loc += s_size;
+  }
+  return *this;
+}
 }  // namespace utt

utt-replica/signature-processor/tests/sigProcessorTests.cpp

@@ -130,6 +130,7 @@ class TestReceiver : public bft::communication::IReceiver {
 struct GpData {
   libutt::CommKey cck;
   libutt::CommKey rck;
+  bool budget_policy = true;
 };
 
 std::tuple<libutt::api::UTTParams, RandSigDKG, RegAuthSK> init(size_t n, size_t thresh) {
@@ -304,9 +305,13 @@ class test_utt_instance : public ::testing::Test {
       msr->registerMsgHandler(MsgCode::ClientRequest, [&, sp](bftEngine::impl::MessageBase* message) {
         ClientRequestMsg* msg = (ClientRequestMsg*)message;
         uint64_t job_id{0};
-        libutt::api::types::Signature fsig(msg->requestLength() - sizeof(uint64_t));
+
+        std::vector<uint8_t> cs_buffer(msg->requestLength() - sizeof(uint64_t));
         std::memcpy(&job_id, msg->requestBuf(), sizeof(uint64_t));
-        std::memcpy(fsig.data(), msg->requestBuf() + sizeof(uint64_t), fsig.size());
+        std::memcpy(cs_buffer.data(), msg->requestBuf() + sizeof(uint64_t), cs_buffer.size());
+        utt::SigProcessor::CompleteSignatureMsg cs_msg(cs_buffer);
+        ASSERT_TRUE(cs_msg.validate());
+        auto& fsig = cs_msg.getFullSig();
         {
           std::unique_lock lk(sigs_lock);
           for (size_t j = 0; j < n; j++) {
@@ -418,7 +423,6 @@ class utt_complete_system : public utt_system_include_budget {
   }
   std::unordered_map<std::string, std::vector<libutt::api::Coin>> coins;
 };
-
 TEST_F(test_utt_instance, test_clients_registration) {
   registerClients(
       job_id,

utt/CMakeLists.txt

@@ -41,6 +41,12 @@ option(
   OFF
 )
 
+option(
+  BUILD_PRIVACY_ADMIN_CLI
+  "Enable building of admin-cli"
+  OFF
+)
+
 #
 # Configure CCache if available
 #
@@ -186,6 +192,9 @@ add_subdirectory(libxutils)
 if(BUILD_WALLET_CLI)
     add_subdirectory(wallet-cli) 
 endif()
+if(BUILD_PRIVACY_ADMIN_CLI)
+    add_subdirectory(admin-cli) 
+endif()
 
 # [TODO-UTT] This improved api for libutt could go into its own subproject 
 set(newutt_src 

utt/admin-cli/CMakeLists.txt

@@ -0,0 +1,14 @@
+add_subdirectory("proto")
+
+set(privacy-admin-cli-src
+    src/main.cpp
+    src/admin.cpp
+)
+
+add_executable(privacy-admin-cli ${privacy-admin-cli-src})
+
+target_include_directories(privacy-admin-cli PUBLIC include/ ../utt-client-api/include ../utt-common-api/include)
+
+target_link_libraries(privacy-admin-cli PUBLIC
+  privacy-admin-api-proto utt_client_api
+)

utt/admin-cli/README.md

@@ -0,0 +1,3 @@
+# UTT Admin Command-Line Application
+
+A command-line application for a UTT admin. Uses gRPC to talk to a admin service, for which we provide the interface in [proto/api/v1/api.proto](proto/api/v1/api.proto). The actual implementation of an admin service is not provided.