fix: ban peer if it sends bad liveness data (#5844)

Description
---
Bans the peer if they sent invalid liveness data.

Motivation and Context
---
I think there was a change to the chainmetadata struct recently, and
this resulted in a bunch of warnings on my igor node. This PR now bans
the node if they sent data that doesn't deserialize.

How Has This Been Tested?
---
Tested with igor trying to connect to old nodes

What process can a PR reviewer use to test or verify this change?
---
Try to connect to an igor node running 0.52.0-pre.1

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->

base_layer/core/src/base_node/chain_metadata_service/initializer.rs

@@ -21,6 +21,7 @@
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use log::*;
+use tari_comms::connectivity::ConnectivityRequester;
 use tari_p2p::services::liveness::LivenessHandle;
 use tari_service_framework::{async_trait, ServiceInitializationError, ServiceInitializer, ServiceInitializerContext};
 use tokio::sync::broadcast;
@@ -42,10 +43,11 @@ impl ServiceInitializer for ChainMetadataServiceInitializer {
         context.register_handle(handle);
 
         context.spawn_until_shutdown(|handles| {
+            let connectivity = handles.expect_handle::<ConnectivityRequester>();
             let liveness = handles.expect_handle::<LivenessHandle>();
             let base_node = handles.expect_handle::<LocalNodeCommsInterface>();
 
-            ChainMetadataService::new(liveness, base_node, publisher).run()
+            ChainMetadataService::new(liveness, base_node, connectivity, publisher).run()
         });
 
         debug!(target: LOG_TARGET, "Chain Metadata Service initialized");

base_layer/core/src/base_node/chain_metadata_service/service.rs

@@ -27,7 +27,7 @@ use num_format::{Locale, ToFormattedString};
 use prost::Message;
 use tari_common::log_if_error;
 use tari_common_types::chain_metadata::ChainMetadata;
-use tari_comms::message::MessageExt;
+use tari_comms::{connectivity::ConnectivityRequester, message::MessageExt, BAN_DURATION_LONG};
 use tari_p2p::services::liveness::{LivenessEvent, LivenessHandle, MetadataKey, PingPongEvent};
 use tokio::sync::broadcast;
 
@@ -46,6 +46,7 @@ const NUM_ROUNDS_NETWORK_SILENCE: u16 = 3;
 pub(super) struct ChainMetadataService {
     liveness: LivenessHandle,
     base_node: LocalNodeCommsInterface,
+    connectivity: ConnectivityRequester,
     event_publisher: broadcast::Sender<Arc<ChainMetadataEvent>>,
     number_of_rounds_no_pings: u16,
 }
@@ -60,12 +61,14 @@ impl ChainMetadataService {
     pub fn new(
         liveness: LivenessHandle,
         base_node: LocalNodeCommsInterface,
+        connectivity: ConnectivityRequester,
         event_publisher: broadcast::Sender<Arc<ChainMetadataEvent>>,
     ) -> Self {
         Self {
             liveness,
             base_node,
             event_publisher,
+            connectivity,
             number_of_rounds_no_pings: 0,
         }
     }
@@ -85,21 +88,28 @@ impl ChainMetadataService {
             tokio::select! {
                 Ok(block_event) = block_event_stream.recv() => {
                     log_if_error!(
-                        level: debug,
+                        level: info,
                         target: LOG_TARGET,
                         "Failed to handle block event because '{}'",
                         self.handle_block_event(&block_event).await
                     );
                 },
 
                 Ok(event) = liveness_event_stream.recv() => {
-                    log_if_error!(
-                        target: LOG_TARGET,
-                        "Failed to handle liveness event because '{}'",
-                        self.handle_liveness_event(&event).await
-                    );
-                },
+                    match
+                        self.handle_liveness_event(&event).await {
+                        Ok(_) => {}
+                        Err(e) => {
+                           info!( target: LOG_TARGET, "Failed to handle liveness event because '{}'", e);
+                           if let ChainMetadataSyncError::ReceivedInvalidChainMetadata(node_id,reason) = e {
+                               log_if_error!(
+                                 level: info,
+                                 target: LOG_TARGET, "Failed to ban node '{}'",
+                                 self.connectivity.ban_peer_until(node_id, BAN_DURATION_LONG, reason).await);                                           }
+                        }
+                    }
 
+                },
             }
         }
     }
@@ -215,7 +225,7 @@ mod test {
     use std::convert::TryInto;
 
     use futures::StreamExt;
-    use tari_comms::peer_manager::NodeId;
+    use tari_comms::{peer_manager::NodeId, test_utils::mocks::create_connectivity_mock};
     use tari_p2p::services::liveness::{
         mock::{create_p2p_liveness_mock, LivenessMockState},
         LivenessRequest,
@@ -268,7 +278,9 @@ mod test {
         let (base_node, base_node_receiver) = create_base_node_nci();
         let (publisher, event_rx) = broadcast::channel(10);
 
-        let service = ChainMetadataService::new(liveness_handle, base_node, publisher);
+        let connectivity = create_connectivity_mock();
+
+        let service = ChainMetadataService::new(liveness_handle, base_node, connectivity.0, publisher);
 
         (service, liveness_mock_state, base_node_receiver, event_rx)
     }

comms/core/src/lib.rs

@@ -52,6 +52,7 @@ pub mod utils;
 pub mod peer_validator;
 
 mod bans;
+pub use bans::{BAN_DURATION_LONG, BAN_DURATION_SHORT};
 pub mod test_utils;
 pub mod traits;