[Consensus] add BlockVerifier (#16147)

## Description 

- Add `BlockVerifier` trait and `SignedBlockVerifier` impl for verifying
block metadata, transactions and ancestors.
- Add `TransactionVerifier` trait, to be implemented by
`SuiTxValidator`.
- Use `ProtocolKeyPair` instead of `NetworkKeyPair` or generic type for
block signing. `ProtocolKeyPair` is currently BLS and can switch to
EdDSA later.
- Include block signature when computing `BlockDigest`.

## Test Plan 

Unit tests.

---
If your changes are not user-facing and do not break anything, you can
skip the following section. Otherwise, please briefly describe what has
changed under the Release Notes section.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

consensus/config/src/committee.rs

@@ -56,7 +56,8 @@ impl Committee {
         }
     }
 
-    /// Public accessors for Committee fields.
+    /// -----------------------------------------------------------------------
+    /// Accessors to Committee fields.
 
     pub fn epoch(&self) -> Epoch {
         self.epoch
@@ -74,20 +75,36 @@ impl Committee {
         self.validity_threshold
     }
 
-    /// Returns true if the provided stake has reached quorum (2f+1)
+    pub fn stake(&self, authority_index: AuthorityIndex) -> Stake {
+        self.authorities[authority_index].stake
+    }
+
+    pub fn authority(&self, authority_index: AuthorityIndex) -> &Authority {
+        &self.authorities[authority_index]
+    }
+
+    pub fn authorities(&self) -> impl Iterator<Item = (AuthorityIndex, &Authority)> {
+        self.authorities
+            .iter()
+            .enumerate()
+            .map(|(i, a)| (AuthorityIndex(i as u32), a))
+    }
+
+    /// -----------------------------------------------------------------------
+    /// Helpers for Committee properties.
+
+    /// Returns true if the provided stake has reached quorum (2f+1).
     pub fn reached_quorum(&self, stake: Stake) -> bool {
         stake >= self.quorum_threshold()
     }
 
-    /// Returns true if the provided stake has reached validity (f+1)
+    /// Returns true if the provided stake has reached validity (f+1).
     pub fn reached_validity(&self, stake: Stake) -> bool {
         stake >= self.validity_threshold()
     }
 
-    pub fn stake(&self, authority_index: AuthorityIndex) -> Stake {
-        self.authorities[authority_index].stake
-    }
-
+    /// Coverts an index to an AuthorityIndex, if valid.
+    /// Returns None if index is out of bound.
     pub fn to_authority_index(&self, index: usize) -> Option<AuthorityIndex> {
         if index < self.authorities.len() {
             Some(AuthorityIndex(index as u32))
@@ -96,21 +113,12 @@ impl Committee {
         }
     }
 
-    pub fn authority(&self, authority_index: AuthorityIndex) -> &Authority {
-        &self.authorities[authority_index]
-    }
-
-    pub fn authorities(&self) -> impl Iterator<Item = (AuthorityIndex, &Authority)> {
-        self.authorities
-            .iter()
-            .enumerate()
-            .map(|(i, a)| (AuthorityIndex(i as u32), a))
-    }
-
-    pub fn exists(&self, index: AuthorityIndex) -> bool {
+    /// Returns true if the provided index is valid.
+    pub fn is_valid_index(&self, index: AuthorityIndex) -> bool {
         index.value() < self.size()
     }
 
+    /// Returns number of authorities in the committee.
     pub fn size(&self) -> usize {
         self.authorities.len()
     }

consensus/config/src/crypto.rs

@@ -17,18 +17,16 @@ use shared_crypto::intent::INTENT_PREFIX_LENGTH;
 // to change all four aliases to point to concrete types that work with each other. Failure to do
 // so will result in a ton of compilation errors, and worse: it will not make sense!
 
-/// Network key signs network messages and blocks.
+/// Network key is used for TLS and network identity of the authority.
 pub type NetworkPublicKey = ed25519::Ed25519PublicKey;
 pub type NetworkPrivateKey = ed25519::Ed25519PrivateKey;
 pub type NetworkKeyPair = ed25519::Ed25519KeyPair;
-pub type NetworkKeySignature = ed25519::Ed25519Signature;
-pub type NetworkKeySignatureAsBytes = ed25519::Ed25519SignatureAsBytes;
 
-/// Protocol key is used in random beacon.
+/// Protocol key is used for signing blocks and verifying block signatures.
 pub type ProtocolPublicKey = bls12381::min_sig::BLS12381PublicKey;
-pub type ProtocolPublicKeyBytes = bls12381::min_sig::BLS12381PublicKeyAsBytes;
 pub type ProtocolPrivateKey = bls12381::min_sig::BLS12381PrivateKey;
 pub type ProtocolKeyPair = bls12381::min_sig::BLS12381KeyPair;
+pub type ProtocolKeySignature = bls12381::min_sig::BLS12381Signature;
 
 /// For block digest.
 pub type DefaultHashFunction = Blake2b256;

consensus/core/src/authority_node.rs

@@ -3,42 +3,59 @@
 
 use std::sync::Arc;
 use std::time::Instant;
+use std::vec;
 
+use async_trait::async_trait;
+use bytes::Bytes;
 use consensus_config::{AuthorityIndex, Committee, NetworkKeyPair, Parameters, ProtocolKeyPair};
 use parking_lot::RwLock;
 use prometheus::Registry;
 use sui_protocol_config::ProtocolConfig;
 use tracing::info;
 
+use crate::block::{BlockAPI, BlockRef, SignedBlock, VerifiedBlock};
 use crate::block_manager::BlockManager;
-use crate::block_verifier::BlockVerifier;
+use crate::block_verifier::{BlockVerifier, SignedBlockVerifier};
 use crate::context::Context;
 use crate::core::{Core, CoreSignals};
-use crate::core_thread::CoreThreadDispatcher;
+use crate::core_thread::{ChannelCoreThreadDispatcher, CoreThreadDispatcher, CoreThreadHandle};
 use crate::dag_state::DagState;
+use crate::error::{ConsensusError, ConsensusResult};
 use crate::leader_timeout::{LeaderTimeoutTask, LeaderTimeoutTaskHandle};
 use crate::metrics::initialise_metrics;
+use crate::network::{NetworkManager, NetworkService};
 use crate::storage::rocksdb_store::RocksDBStore;
-use crate::transactions_client::{TransactionsClient, TransactionsConsumer};
+use crate::transaction::{TransactionClient, TransactionConsumer, TransactionVerifier};
 
-pub struct AuthorityNode {
+pub struct AuthorityNode<N>
+where
+    N: NetworkManager<AuthorityService>,
+{
     context: Arc<Context>,
     start_time: Instant,
-    transactions_client: Arc<TransactionsClient>,
+    transaction_client: Arc<TransactionClient>,
     leader_timeout_handle: LeaderTimeoutTaskHandle,
+    // Keeps core thread running.
+    core_thread_handle: CoreThreadHandle,
+    // Keeps network alive.
+    network_manager: N,
 }
 
-impl AuthorityNode {
+impl<N> AuthorityNode<N>
+where
+    N: NetworkManager<AuthorityService>,
+{
     #[allow(unused)]
     async fn start(
         own_index: AuthorityIndex,
         committee: Committee,
         parameters: Parameters,
         protocol_config: ProtocolConfig,
-        // To avoid accidentally leaking the private key, the key pair should only be stored in core
-        block_signer: NetworkKeyPair,
-        _signer: ProtocolKeyPair,
-        _block_verifier: impl BlockVerifier,
+        // To avoid accidentally leaking the private key, the protocol key pair should only be
+        // kept in Core.
+        protocol_keypair: ProtocolKeyPair,
+        network_keypair: NetworkKeyPair,
+        transaction_verifier: Arc<dyn TransactionVerifier>,
         registry: Registry,
     ) -> Self {
         info!("Starting authority with index {}", own_index);
@@ -52,10 +69,14 @@ impl AuthorityNode {
         let start_time = Instant::now();
 
         // Create the transactions client and the transactions consumer
-        let (client, tx_receiver) = TransactionsClient::new(context.clone());
-        let tx_consumer = TransactionsConsumer::new(tx_receiver, context.clone(), None);
+        let (client, tx_receiver) = TransactionClient::new(context.clone());
+        let tx_consumer = TransactionConsumer::new(tx_receiver, context.clone(), None);
 
-        // Construct Core
+        // Create network manager and client.
+        let network_manager = N::new(context.clone());
+        let _client = network_manager.client();
+
+        // Construct Core components.
         let (core_signals, signals_receivers) = CoreSignals::new();
         let store = Arc::new(RocksDBStore::new(&context.parameters.db_path_str_unsafe()));
         let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store.clone())));
@@ -65,20 +86,34 @@ impl AuthorityNode {
             tx_consumer,
             block_manager,
             core_signals,
-            block_signer,
+            protocol_keypair,
             store,
         );
 
-        let (core_dispatcher, core_dispatcher_handle) =
-            CoreThreadDispatcher::start(core, context.clone());
+        let (core_dispatcher, core_thread_handle) =
+            ChannelCoreThreadDispatcher::start(core, context.clone());
         let leader_timeout_handle =
-            LeaderTimeoutTask::start(core_dispatcher, signals_receivers, context.clone());
+            LeaderTimeoutTask::start(core_dispatcher.clone(), signals_receivers, context.clone());
+
+        // Start network service.
+        let block_verifier = Arc::new(SignedBlockVerifier::new(
+            context.clone(),
+            transaction_verifier,
+        ));
+        let network_service = Arc::new(AuthorityService {
+            context: context.clone(),
+            block_verifier,
+            core_dispatcher,
+        });
+        network_manager.install_service(network_keypair, network_service);
 
         Self {
             context,
             start_time,
+            transaction_client: Arc::new(client),
             leader_timeout_handle,
-            transactions_client: Arc::new(client),
+            core_thread_handle,
+            network_manager,
         }
     }
 
@@ -89,6 +124,8 @@ impl AuthorityNode {
             self.start_time.elapsed()
         );
 
+        self.network_manager.stop().await;
+        self.core_thread_handle.stop();
         self.leader_timeout_handle.stop().await;
 
         self.context
@@ -99,45 +136,104 @@ impl AuthorityNode {
     }
 
     #[allow(unused)]
-    pub fn transactions_client(&self) -> Arc<TransactionsClient> {
-        self.transactions_client.clone()
+    pub fn transaction_client(&self) -> Arc<TransactionClient> {
+        self.transaction_client.clone()
+    }
+}
+
+/// Authority's network interface.
+pub struct AuthorityService {
+    context: Arc<Context>,
+    block_verifier: Arc<dyn BlockVerifier>,
+    core_dispatcher: ChannelCoreThreadDispatcher,
+}
+
+#[async_trait]
+impl NetworkService for AuthorityService {
+    async fn handle_send_block(
+        &self,
+        peer: AuthorityIndex,
+        serialized_block: Bytes,
+    ) -> ConsensusResult<()> {
+        // TODO: dedup block verifications, here and with fetched blocks.
+        let signed_block: SignedBlock =
+            bcs::from_bytes(&serialized_block).map_err(ConsensusError::MalformedBlock)?;
+        if peer != signed_block.author() {
+            self.context
+                .metrics
+                .node_metrics
+                .invalid_blocks
+                .with_label_values(&[&peer.to_string()])
+                .inc();
+            let e = ConsensusError::UnexpectedAuthority(signed_block.author(), peer);
+            info!("Block with wrong authority from {}: {}", peer, e);
+            return Err(e);
+        }
+        if let Err(e) = self.block_verifier.verify(&signed_block) {
+            self.context
+                .metrics
+                .node_metrics
+                .invalid_blocks
+                .with_label_values(&[&peer.to_string()])
+                .inc();
+            info!("Invalid block from {}: {}", peer, e);
+            return Err(e);
+        }
+        let verified_block = VerifiedBlock::new_verified(signed_block, serialized_block);
+        self.core_dispatcher
+            .add_blocks(vec![verified_block])
+            .await
+            .map_err(|_| ConsensusError::Shutdown)?;
+        Ok(())
+    }
+
+    async fn handle_fetch_blocks(
+        &self,
+        _peer: AuthorityIndex,
+        _block_refs: Vec<BlockRef>,
+    ) -> ConsensusResult<Vec<Bytes>> {
+        Ok(vec![])
     }
 }
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
     use consensus_config::{local_committee_and_keys, NetworkKeyPair, Parameters, ProtocolKeyPair};
     use fastcrypto::traits::ToFromBytes;
     use prometheus::Registry;
     use sui_protocol_config::ProtocolConfig;
     use tempfile::TempDir;
 
     use crate::authority_node::AuthorityNode;
-    use crate::block_verifier::TestBlockVerifier;
+    use crate::network::anemo_network::AnemoManager;
+    use crate::transaction::NoopTransactionVerifier;
 
     #[tokio::test]
     async fn start_and_stop() {
         let (committee, keypairs) = local_committee_and_keys(0, vec![1]);
         let registry = Registry::new();
+
         let temp_dir = TempDir::new().unwrap();
         let parameters = Parameters {
             db_path: Some(temp_dir.into_path()),
             ..Default::default()
         };
-        let block_verifier = TestBlockVerifier {};
+        let txn_verifier = NoopTransactionVerifier {};
 
         let (own_index, _) = committee.authorities().last().unwrap();
-        let block_signer = NetworkKeyPair::from_bytes(keypairs[0].0.as_bytes()).unwrap();
-        let signer = ProtocolKeyPair::from_bytes(keypairs[0].1.as_bytes()).unwrap();
+        let protocol_keypair = ProtocolKeyPair::from_bytes(keypairs[0].1.as_bytes()).unwrap();
+        let network_keypair = NetworkKeyPair::from_bytes(keypairs[0].0.as_bytes()).unwrap();
 
-        let authority = AuthorityNode::start(
+        let authority = AuthorityNode::<AnemoManager>::start(
             own_index,
             committee,
             parameters,
             ProtocolConfig::get_for_max_version_UNSAFE(),
-            block_signer,
-            signer,
-            block_verifier,
+            protocol_keypair,
+            network_keypair,
+            Arc::new(txn_verifier),
             registry,
         )
         .await;