fix: miner delay attack (tari-project#5582)

Description
---
Stops the node blocking until it has a full block

Audit Finding Number
---
TARI-0001, TARI-0002

Motivation and Context
---
Its possible for a malicious node to block access to the full block by
using first the mempool to say it does not have the missing
transactions, then when asked for the full block to only then provide
the full block. But these requests can be made slow to delay the node
constructing the new block by almost 2 mins. This is the block time.

By only accepting 1 block request at a time, a malicious node can lock
down the local node for that entire 2 mins knowing they wont accept any
other blocks. This changes it so that nodes can still process new
requests. The first complete block will be served first.


Also changes the display of block information to ensure block sorted is
correctly used.

---------

Co-authored-by: Stan Bondi <[email protected]>

base_layer/core/src/base_node/comms_interface/inbound_handlers.rs

@@ -21,6 +21,7 @@
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use std::{
+    collections::HashSet,
     convert::{TryFrom, TryInto},
     sync::Arc,
     time::{Duration, Instant},
@@ -31,7 +32,7 @@ use strum_macros::Display;
 use tari_common_types::types::{BlockHash, FixedHash, HashOutput};
 use tari_comms::{connectivity::ConnectivityRequester, peer_manager::NodeId};
 use tari_utilities::hex::Hex;
-use tokio::sync::Semaphore;
+use tokio::sync::RwLock;
 
 use crate::{
     base_node::{
@@ -81,7 +82,7 @@ pub struct InboundNodeCommsHandlers<B> {
     blockchain_db: AsyncBlockchainDb<B>,
     mempool: Mempool,
     consensus_manager: ConsensusManager,
-    new_block_request_semaphore: Arc<Semaphore>,
+    list_of_reconciling_blocks: Arc<RwLock<HashSet<HashOutput>>>,
     outbound_nci: OutboundNodeCommsInterface,
     connectivity: ConnectivityRequester,
 }
@@ -103,7 +104,7 @@ where B: BlockchainBackend + 'static
             blockchain_db,
             mempool,
             consensus_manager,
-            new_block_request_semaphore: Arc::new(Semaphore::new(1)),
+            list_of_reconciling_blocks: Arc::new(RwLock::new(HashSet::new())),
             outbound_nci,
             connectivity,
         }
@@ -436,14 +437,7 @@ where B: BlockchainBackend + 'static
             return Ok(());
         }
 
-        // Only a single block request can complete at a time.
-        // As multiple NewBlock requests arrive from propagation, this semaphore prevents multiple requests to nodes for
-        // the same full block. The first request that succeeds will stop the node from requesting the block from any
-        // other node (block_exists is true).
-        // Arc clone to satisfy the borrow checker
-        let semaphore = self.new_block_request_semaphore.clone();
-        let _permit = semaphore.acquire().await.unwrap();
-
+        // Lets check if the block exists before we try and ask for a complete block
         if self.blockchain_db.block_exists(block_hash).await? {
             debug!(
                 target: LOG_TARGET,
@@ -453,6 +447,39 @@ where B: BlockchainBackend + 'static
             return Ok(());
         }
 
+        {
+            // we use a double lock to make sure we can only reconcile one unique block at a time. We may receive the
+            // same block from multiple peer near simultaneously. We should only reconcile each unique block once.
+            let read_lock = self.list_of_reconciling_blocks.read().await;
+            if read_lock.contains(&block_hash) {
+                debug!(
+                    target: LOG_TARGET,
+                    "Block with hash `{}` is already being reconciled",
+                    block_hash.to_hex()
+                );
+                return Ok(());
+            }
+        }
+        {
+            let mut write_lock = self.list_of_reconciling_blocks.write().await;
+            if self.blockchain_db.block_exists(block_hash).await? {
+                debug!(
+                    target: LOG_TARGET,
+                    "Block with hash `{}` already stored",
+                    block_hash.to_hex()
+                );
+                return Ok(());
+            }
+            if !write_lock.insert(block_hash) {
+                debug!(
+                    target: LOG_TARGET,
+                    "Block with hash `{}` is already being reconciled",
+                    block_hash.to_hex()
+                );
+                return Ok(());
+            }
+        }
+
         debug!(
             target: LOG_TARGET,
             "Block with hash `{}` is unknown. Constructing block from known mempool transactions / requesting missing \
@@ -461,9 +488,23 @@ where B: BlockchainBackend + 'static
             source_peer
         );
 
+        let result = self.reconcile_and_add_block(source_peer.clone(), new_block).await;
+
+        {
+            let mut write_lock = self.list_of_reconciling_blocks.write().await;
+            write_lock.remove(&block_hash);
+        }
+        result?;
+        Ok(())
+    }
+
+    async fn reconcile_and_add_block(
+        &mut self,
+        source_peer: NodeId,
+        new_block: NewBlock,
+    ) -> Result<(), CommsInterfaceError> {
         let block = self.reconcile_block(source_peer.clone(), new_block).await?;
         self.handle_block(block, Some(source_peer)).await?;
-
         Ok(())
     }
 
@@ -890,7 +931,7 @@ impl<B> Clone for InboundNodeCommsHandlers<B> {
             blockchain_db: self.blockchain_db.clone(),
             mempool: self.mempool.clone(),
             consensus_manager: self.consensus_manager.clone(),
-            new_block_request_semaphore: self.new_block_request_semaphore.clone(),
+            list_of_reconciling_blocks: self.list_of_reconciling_blocks.clone(),
             outbound_nci: self.outbound_nci.clone(),
             connectivity: self.connectivity.clone(),
         }