Merge pull request #1325 from KitHat/master

ST-623 -- Init Commit multi-sig process

Author: ashcherbakov

plenum/bls/bls_bft_factory.py

@@ -26,6 +26,7 @@ def create_bls_bft_replica(self, is_master) -> BlsBftReplica:
         return BlsBftReplicaPlenum(self._node.name,
                                    self._node.bls_bft,
                                    is_master,
+                                   self._node.db_manager,
                                    self._node.metrics)
 
 

plenum/bls/bls_bft_replica_plenum.py

@@ -4,11 +4,13 @@
 from crypto.bls.bls_bft import BlsBft
 from crypto.bls.bls_bft_replica import BlsBftReplica
 from crypto.bls.bls_multi_signature import MultiSignature, MultiSignatureValue
-from plenum.common.constants import DOMAIN_LEDGER_ID, BLS_PREFIX, POOL_LEDGER_ID
+from plenum.common.constants import DOMAIN_LEDGER_ID, BLS_PREFIX, POOL_LEDGER_ID, AUDIT_LEDGER_ID, TXN_PAYLOAD, \
+    TXN_PAYLOAD_DATA, AUDIT_TXN_LEDGER_ROOT, AUDIT_TXN_STATE_ROOT, AUDIT_TXN_PP_SEQ_NO
 from plenum.common.messages.node_messages import PrePrepare, Prepare, Commit
 from plenum.common.metrics_collector import MetricsCollector, NullMetricsCollector, measure_time, MetricsName
 from plenum.common.types import f
 from plenum.common.util import compare_3PC_keys
+from plenum.server.database_manager import DatabaseManager
 from stp_core.common.log import getlogger
 
 logger = getlogger()
@@ -19,10 +21,14 @@ def __init__(self,
                  node_id,
                  bls_bft: BlsBft,
                  is_master,
+                 database_manager: DatabaseManager,
                  metrics: MetricsCollector = NullMetricsCollector()):
         super().__init__(bls_bft, is_master)
+        self._all_bls_latest_multi_sigs = None
         self.node_id = node_id
+        self._database_manager = database_manager
         self._signatures = {}
+        self._all_signatures = {}
         self._bls_latest_multi_sig = None  # MultiSignature
         self.state_root_serializer = state_roots_serializer
         self.metrics = metrics
@@ -35,6 +41,13 @@ def _can_process_ledger(self, ledger_id):
 
     @measure_time(MetricsName.BLS_VALIDATE_PREPREPARE_TIME)
     def validate_pre_prepare(self, pre_prepare: PrePrepare, sender):
+        if f.BLS_MULTI_SIGS.nm in pre_prepare and pre_prepare.blsMultiSigs:
+            multi_sigs = pre_prepare.blsMultiSigs
+            for sig in multi_sigs:
+                multi_sig = MultiSignature.from_list(*sig)
+                if not self._validate_multi_sig(multi_sig):
+                    return BlsBftReplica.PPR_BLS_MULTISIG_WRONG
+
         if f.BLS_MULTI_SIG.nm not in pre_prepare or \
                 pre_prepare.blsMultiSig is None:
             return
@@ -48,6 +61,22 @@ def validate_prepare(self, prepare: Prepare, sender):
 
     @measure_time(MetricsName.BLS_VALIDATE_COMMIT_TIME)
     def validate_commit(self, commit: Commit, sender, pre_prepare: PrePrepare):
+        if f.BLS_SIGS.nm in commit:
+            audit_txn = self._get_correct_audit_transaction(pre_prepare)
+            if audit_txn:
+                audit_payload = audit_txn[TXN_PAYLOAD][TXN_PAYLOAD_DATA]
+                for lid, sig in commit.blsSigs.items():
+                    lid = int(lid)
+                    if lid not in audit_payload[AUDIT_TXN_STATE_ROOT] or lid not in audit_payload[AUDIT_TXN_LEDGER_ROOT]:
+                        return BlsBftReplicaPlenum.CM_BLS_SIG_WRONG
+                    if not self._validate_signature(sender, sig,
+                                                    BlsBftReplicaPlenum._create_fake_pre_prepare_for_multi_sig(
+                                                        lid,
+                                                        audit_payload[AUDIT_TXN_STATE_ROOT][lid],
+                                                        audit_payload[AUDIT_TXN_LEDGER_ROOT][lid],
+                                                        pre_prepare
+                                                    )):
+                        return BlsBftReplicaPlenum.CM_BLS_SIG_WRONG
         if f.BLS_SIG.nm not in commit:
             # TODO: It's optional for now
             return
@@ -62,12 +91,14 @@ def update_pre_prepare(self, pre_prepare_params, ledger_id):
         if not self._can_process_ledger(ledger_id):
             return pre_prepare_params
 
-        if not self._bls_latest_multi_sig:
-            return pre_prepare_params
+        if self._bls_latest_multi_sig is not None:
+            pre_prepare_params.append(self._bls_latest_multi_sig.as_list())
+            self._bls_latest_multi_sig = None
 
-        pre_prepare_params.append(self._bls_latest_multi_sig.as_list())
-        self._bls_latest_multi_sig = None
         # Send signature in COMMITs only
+        if self._all_bls_latest_multi_sigs is not None:
+            pre_prepare_params.append([val.as_list() for val in self._all_bls_latest_multi_sigs])
+            self._all_bls_latest_multi_sigs = None
 
         return pre_prepare_params
 
@@ -92,6 +123,22 @@ def update_commit(self, commit_params, pre_prepare: PrePrepare):
         logger.debug("{}{} signed COMMIT {} for state {} with sig {}"
                      .format(BLS_PREFIX, self, commit_params, state_root_hash, bls_signature))
         commit_params.append(bls_signature)
+
+        last_audit_txn = self._get_correct_audit_transaction(pre_prepare)
+        if last_audit_txn:
+            res = {}
+            payload_data = last_audit_txn[TXN_PAYLOAD][TXN_PAYLOAD_DATA]
+            for ledger_id in payload_data[AUDIT_TXN_STATE_ROOT].keys():
+                fake_pp = BlsBftReplicaPlenum._create_fake_pre_prepare_for_multi_sig(
+                    ledger_id,
+                    payload_data[AUDIT_TXN_STATE_ROOT].get(ledger_id),
+                    payload_data[AUDIT_TXN_LEDGER_ROOT].get(ledger_id),
+                    pre_prepare
+                )
+                bls_signature = self._sign_state(fake_pp)
+                res[str(ledger_id)] = bls_signature
+            commit_params.append(res)
+
         return commit_params
 
     # ----PROCESS----
@@ -105,15 +152,19 @@ def process_prepare(self, prepare: Prepare, sender):
         pass
 
     def process_commit(self, commit: Commit, sender):
-        if f.BLS_SIG.nm not in commit:
-            return
-        if commit.blsSig is None:
-            return
-
         key_3PC = (commit.viewNo, commit.ppSeqNo)
-        if key_3PC not in self._signatures:
-            self._signatures[key_3PC] = {}
-        self._signatures[key_3PC][self.get_node_name(sender)] = commit.blsSig
+        if f.BLS_SIG.nm in commit and commit.blsSig is not None:
+            if key_3PC not in self._signatures:
+                self._signatures[key_3PC] = {}
+            self._signatures[key_3PC][self.get_node_name(sender)] = commit.blsSig
+
+        if f.BLS_SIGS.nm in commit and commit.blsSigs is not None:
+            if key_3PC not in self._all_signatures:
+                self._all_signatures[key_3PC] = {}
+            for ledger_id in commit.blsSigs.keys():
+                if ledger_id not in self._all_signatures[key_3PC]:
+                    self._all_signatures[key_3PC][ledger_id] = {}
+                self._all_signatures[key_3PC][ledger_id][self.get_node_name(sender)] = commit.blsSigs[ledger_id]
 
     def process_order(self, key, quorums, pre_prepare):
         if not self._can_process_ledger(pre_prepare.ledgerId):
@@ -126,12 +177,19 @@ def process_order(self, key, quorums, pre_prepare):
         # but save on master only
         bls_multi_sig = self._calculate_multi_sig(key, pre_prepare)
 
+        all_bls_multi_sigs = self._calculate_all_multi_sigs(key, pre_prepare)
+
         if not self._is_master:
             return
 
-        self._save_multi_sig_local(bls_multi_sig)
+        if all_bls_multi_sigs:
+            for bls_multi_sig in all_bls_multi_sigs:
+                self._save_multi_sig_local(bls_multi_sig)
+        elif bls_multi_sig:
+            self._save_multi_sig_local(bls_multi_sig)
 
         self._bls_latest_multi_sig = bls_multi_sig
+        self._all_bls_latest_multi_sigs = all_bls_multi_sigs
 
     # ----GC----
 
@@ -142,6 +200,7 @@ def gc(self, key_3PC):
                 keys_to_remove.append(key)
         for key in keys_to_remove:
             self._signatures.pop(key, None)
+            self._all_signatures.pop(key, None)
 
     # ----MULT_SIG----
 
@@ -154,7 +213,7 @@ def _create_multi_sig_value_for_pre_prepare(self, pre_prepare: PrePrepare, pool_
         return multi_sig_value
 
     def validate_key_proof_of_possession(self, key_proof, pk):
-        return self._bls_bft.bls_crypto_verifier\
+        return self._bls_bft.bls_crypto_verifier \
             .verify_key_proof_of_possession(key_proof, pk)
 
     def _validate_signature(self, sender, bls_sig, pre_prepare: PrePrepare):
@@ -198,6 +257,28 @@ def _sign_state(self, pre_prepare: PrePrepare):
     def _can_calculate_multi_sig(self,
                                  key_3PC,
                                  quorums) -> bool:
+        if key_3PC in self._all_signatures:
+            sigs_for_request = self._all_signatures[key_3PC]
+            sigs_invalid = list(
+                filter(
+                    lambda item: not quorums.bls_signatures.is_reached(len(list(item[1].values()))),
+                    sigs_for_request.items()
+                )
+            )
+            if sigs_invalid:
+                for lid, sigs in sigs_invalid:
+                    logger.debug(
+                        '{}Can not create bls signatures for batch {}: '
+                        'There are only {} signatures for ledger {}, '
+                        'while {} required for multi_signature'.format(BLS_PREFIX,
+                                                                       key_3PC,
+                                                                       len(list(sigs.values())),
+                                                                       quorums.bls_signatures.value,
+                                                                       lid)
+                    )
+            else:
+                return True
+
         if key_3PC not in self._signatures:
             return False
 
@@ -216,6 +297,26 @@ def _can_calculate_multi_sig(self,
 
     def _calculate_multi_sig(self, key_3PC, pre_prepare) -> Optional[MultiSignature]:
         sigs_for_request = self._signatures[key_3PC]
+        return self._calculate_single_multi_sig(sigs_for_request, pre_prepare)
+
+    def _calculate_all_multi_sigs(self, key_3PC, pre_prepare) -> Optional[list]:
+        sigs_for_request = self._all_signatures.get(key_3PC)
+        res = []
+        if sigs_for_request:
+            for lid in sigs_for_request:
+                sig = sigs_for_request[lid]
+                audit_txn = self._get_correct_audit_transaction(pre_prepare)
+                if audit_txn:
+                    audit_payload = audit_txn[TXN_PAYLOAD][TXN_PAYLOAD_DATA]
+                    fake_pp = BlsBftReplicaPlenum. \
+                        _create_fake_pre_prepare_for_multi_sig(int(lid),
+                                                               audit_payload[AUDIT_TXN_STATE_ROOT][int(lid)],
+                                                               audit_payload[AUDIT_TXN_LEDGER_ROOT][int(lid)],
+                                                               pre_prepare)
+                    res.append(self._calculate_single_multi_sig(sig, fake_pp))
+        return res
+
+    def _calculate_single_multi_sig(self, sigs_for_request, pre_prepare) -> Optional[MultiSignature]:
         bls_signatures = list(sigs_for_request.values())
         participants = list(sigs_for_request.keys())
 
@@ -245,7 +346,16 @@ def _save_multi_sig_local(self,
 
     def _save_multi_sig_shared(self, pre_prepare: PrePrepare):
 
-        if f.BLS_MULTI_SIG.nm not in pre_prepare:
+        if f.BLS_MULTI_SIGS.nm in pre_prepare and pre_prepare.blsMultiSigs is not None:
+            multi_sigs = pre_prepare.blsMultiSigs
+            for sig in multi_sigs:
+                multi_sig = MultiSignature.from_list(*sig)
+                self._bls_bft.bls_store.put(multi_sig)
+                logger.debug("{}{} saved multi signature {} for root {} (calculated by Primary)"
+                             .format(BLS_PREFIX, self, multi_sig,
+                                     multi_sig.value.state_root_hash))
+            return
+        elif f.BLS_MULTI_SIG.nm not in pre_prepare:
             return
         if pre_prepare.blsMultiSig is None:
             return
@@ -257,6 +367,41 @@ def _save_multi_sig_shared(self, pre_prepare: PrePrepare):
                              multi_sig.value.state_root_hash))
         # TODO: support multiple multi-sigs for multiple previous batches
 
+    def _get_correct_audit_transaction(self, pp: PrePrepare):
+        ledger = self._database_manager.get_ledger(AUDIT_LEDGER_ID)
+        if ledger is None:
+            return None
+        seqNo = ledger.uncommitted_size
+        for curSeqNo in reversed(range(1, seqNo + 1)):
+            txn = ledger.get_by_seq_no_uncommitted(curSeqNo)
+            if txn:
+                payload = txn[TXN_PAYLOAD][TXN_PAYLOAD_DATA]
+                if pp.ppSeqNo == payload[AUDIT_TXN_PP_SEQ_NO]:
+                    return txn
+        return None
+
+    @staticmethod
+    def _create_fake_pre_prepare_for_multi_sig(lid, state_root_hash, txn_root_hash, pre_prepare):
+        params = [
+            pre_prepare.instId,
+            pre_prepare.viewNo,
+            pre_prepare.ppSeqNo,
+            pre_prepare.ppTime,
+            pre_prepare.reqIdr,
+            pre_prepare.discarded,
+            pre_prepare.digest,
+            1,  # doing it to work around the ledgers that are not in plenum -- it will fail the validation of pre-prepare
+            state_root_hash,
+            txn_root_hash,
+            pre_prepare.sub_seq_no,
+            pre_prepare.final,
+            pre_prepare.poolStateRootHash,
+            pre_prepare.auditTxnRootHash
+        ]
+        pp = PrePrepare(*params)
+        pp.ledgerId = lid
+        return pp
+
     @staticmethod
     def get_node_name(replica_name: str):
         # TODO: there is the same method in Replica

plenum/common/messages/node_messages.py

@@ -14,7 +14,7 @@
     SerializedValueField, SignatureField, TieAmongField, AnyValueField, TimestampField, \
     LedgerIdField, MerkleRootField, Base58Field, LedgerInfoField, AnyField, ChooseField, AnyMapField, \
     LimitedLengthStringField, BlsMultiSignatureField, ProtocolVersionField, BooleanField, \
-    IntegerField, BatchIDField, ViewChangeField
+    IntegerField, BatchIDField, ViewChangeField, MapField, StringifiedNonNegativeNumberField
 from plenum.common.messages.message_base import \
     MessageBase
 from plenum.common.types import f
@@ -133,6 +133,8 @@ class PrePrepare(MessageBase):
         # TODO: support multiple multi-sigs for multiple previous batches
         (f.BLS_MULTI_SIG.nm, BlsMultiSignatureField(optional=True,
                                                     nullable=True)),
+        (f.BLS_MULTI_SIGS.nm, IterableField(optional=True,
+                                            inner_field_type=BlsMultiSignatureField(optional=True, nullable=True))),
         (f.ORIGINAL_VIEW_NO.nm, NonNegativeNumberField(optional=True,
                                                        nullable=True)),
         (f.PLUGIN_FIELDS.nm, AnyMapField(optional=True, nullable=True)),
@@ -147,6 +149,13 @@ def _post_process(self, input_as_dict: Dict) -> Dict:
         if bls is not None:
             input_as_dict[f.BLS_MULTI_SIG.nm] = (bls[0], tuple(bls[1]), tuple(bls[2]))
 
+        bls_sigs = input_as_dict.get(f.BLS_MULTI_SIGS.nm, None)
+        if bls_sigs is not None:
+            sub = []
+            for sig in bls_sigs:
+                sub.append((sig[0], tuple(sig[1]), tuple(sig[2])))
+            input_as_dict[f.BLS_MULTI_SIGS.nm] = tuple(sub)
+
         return input_as_dict
 
 
@@ -191,10 +200,12 @@ class Commit(MessageBase):
         (f.PP_SEQ_NO.nm, NonNegativeNumberField()),
         (f.BLS_SIG.nm, LimitedLengthStringField(max_length=BLS_SIG_LIMIT,
                                                 optional=True)),
-
+        (f.BLS_SIGS.nm, MapField(optional=True,
+                                 key_field=StringifiedNonNegativeNumberField(),
+                                 value_field=LimitedLengthStringField(max_length=BLS_SIG_LIMIT))),
         # PLUGIN_FIELDS is not used in Commit as of now but adding for
         # consistency
-        (f.PLUGIN_FIELDS.nm, AnyMapField(optional=True, nullable=True))
+        (f.PLUGIN_FIELDS.nm, AnyMapField(optional=True, nullable=True)),
     )
 
 

plenum/common/types.py

@@ -63,7 +63,9 @@ class f:  # provides a namespace for reusable field constants
     AUDIT_TXN_ROOT_HASH = Field("auditTxnRootHash", str)
     TXN_ROOT = Field("txnRootHash", str)
     BLS_SIG = Field("blsSig", str)
+    BLS_SIGS = Field("blsSigs", Dict[int, str])
     BLS_MULTI_SIG = Field("blsMultiSig", str)
+    BLS_MULTI_SIGS = Field("blsMultiSigs", str)
     BLS_MULTI_SIG_STATE_ROOT = Field("blsMultiSigStateRoot", str)
     MERKLE_ROOT = Field("merkleRoot", str)
     OLD_MERKLE_ROOT = Field("oldMerkleRoot", str)

plenum/server/database_manager.py

@@ -29,7 +29,7 @@ def get_database(self, lid):
             return None
         return self.databases[lid]
 
-    def get_ledger(self, lid):
+    def get_ledger(self, lid) -> Ledger:
         if lid not in self.databases:
             return None
         return self.databases[lid].ledger