chore: enable gossipsub validation

src/libp2p/behaviour.rs

@@ -26,8 +26,9 @@ use ahash::{HashMap, HashSet};
 use libp2p::{
     Multiaddr, allow_block_list, connection_limits,
     gossipsub::{
-        self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId,
-        PublishError, SubscriptionError, ValidationMode, WhitelistSubscriptionFilter,
+        self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAcceptance,
+        MessageAuthenticity, MessageId, PublishError, SubscriptionError, ValidationMode,
+        WhitelistSubscriptionFilter,
     },
     identity::{Keypair, PeerId},
     kad::QueryId,
@@ -100,6 +101,8 @@ pub(in crate::libp2p) fn build_gossipsub(
     let mut gs_config_builder = gossipsub::ConfigBuilder::default();
     gs_config_builder.max_transmit_size(1 << 20);
     gs_config_builder.validation_mode(ValidationMode::Strict);
+    // Forward messages only once validated.
+    gs_config_builder.validate_messages();
     gs_config_builder.message_id_fn(|msg: &gossipsub::Message| {
         let s = blake2b_256(&msg.data);
         MessageId::from(s)
@@ -237,6 +240,17 @@ impl ForestBehaviour {
         self.discovery.peers()
     }
 
+    /// Reports a `gossipsub` message's verdict, which decides forwarding and peer scoring.
+    pub fn report_message_validation_result(
+        &mut self,
+        message_id: &MessageId,
+        source: &PeerId,
+        acceptance: MessageAcceptance,
+    ) {
+        self.gossipsub
+            .report_message_validation_result(message_id, source, acceptance);
+    }
+
     /// Returns a map of peer ids and their multi-addresses
     pub fn peer_addresses(&self) -> HashMap<PeerId, HashSet<Multiaddr>> {
         self.discovery.peer_addresses()

src/libp2p/mod.rs

@@ -25,5 +25,6 @@ pub use self::{config::*, peer_manager::*, service::*};
 #[cfg(test)]
 mod tests {
     mod decode_test;
+    mod gossip_validation_test;
     mod gossipsub_filter_test;
 }

src/libp2p/service.rs

@@ -16,6 +16,7 @@ use anyhow::Context as _;
 use flume::Sender;
 use futures::{select, stream::StreamExt as _};
 pub use libp2p::gossipsub::{IdentTopic, Topic};
+use libp2p::gossipsub::{MessageAcceptance, MessageId};
 use libp2p::{
     PeerId, Swarm, SwarmBuilder,
     autonat::NatStatus,
@@ -651,55 +652,76 @@ async fn handle_discovery_event(
     }
 }
 
-async fn handle_gossip_event(
+/// Verdict for a gossiped message.
+#[derive(Debug)]
+pub(in crate::libp2p) struct Verdict {
+    pub(in crate::libp2p) message_id: MessageId,
+    pub(in crate::libp2p) source: PeerId,
+    pub(in crate::libp2p) acceptance: MessageAcceptance,
+}
+
+/// Decodes a `gossipsub` message, emits it and returns its verdict.
+pub(in crate::libp2p) async fn handle_gossip_event(
     e: gossipsub::Event,
     network_sender_out: &Sender<NetworkEvent>,
     pubsub_block_str: &str,
     pubsub_msg_str: &str,
-) {
-    if let gossipsub::Event::Message {
+) -> Option<Verdict> {
+    let gossipsub::Event::Message {
         propagation_source: source,
+        message_id,
         message,
-        ..
     } = e
-    {
-        let topic = message.topic.as_str();
-        let message = message.data;
-        trace!("Got a Gossip Message from {:?}", source);
-        if topic == pubsub_block_str {
-            match from_slice_with_fallback::<GossipBlock>(&message) {
-                Ok(b) => {
-                    emit_event(
-                        network_sender_out,
-                        NetworkEvent::PubsubMessage {
-                            message: PubsubMessage::Block(b),
-                        },
-                    )
-                    .await;
-                }
-                Err(e) => {
-                    warn!("Gossip Block from peer {source:?} could not be deserialized: {e:#}",);
-                }
+    else {
+        return None;
+    };
+    let topic = message.topic.as_str();
+    let message = message.data;
+
+    trace!("Got a Gossip Message from {:?}", source);
+    let acceptance = if topic == pubsub_block_str {
+        match from_slice_with_fallback::<GossipBlock>(&message) {
+            Ok(b) => {
+                emit_event(
+                    network_sender_out,
+                    NetworkEvent::PubsubMessage {
+                        message: PubsubMessage::Block(b),
+                    },
+                )
+                .await;
+                MessageAcceptance::Accept
             }
-        } else if topic == pubsub_msg_str {
-            match from_slice_with_fallback::<SignedMessage>(&message) {
-                Ok(m) => {
-                    emit_event(
-                        network_sender_out,
-                        NetworkEvent::PubsubMessage {
-                            message: PubsubMessage::Message(m),
-                        },
-                    )
-                    .await;
-                }
-                Err(e) => {
-                    warn!("Gossip Message from peer {source:?} could not be deserialized: {e:#}");
-                }
+            Err(e) => {
+                warn!("Gossip Block from peer {source:?} could not be deserialized: {e:#}",);
+                MessageAcceptance::Ignore
             }
-        } else {
-            warn!("Getting gossip messages from unknown topic: {topic}");
         }
-    }
+    } else if topic == pubsub_msg_str {
+        match from_slice_with_fallback::<SignedMessage>(&message) {
+            Ok(m) => {
+                emit_event(
+                    network_sender_out,
+                    NetworkEvent::PubsubMessage {
+                        message: PubsubMessage::Message(m),
+                    },
+                )
+                .await;
+                MessageAcceptance::Accept
+            }
+            Err(e) => {
+                warn!("Gossip Message from peer {source:?} could not be deserialized: {e:#}");
+                MessageAcceptance::Ignore
+            }
+        }
+    } else {
+        warn!("Getting gossip messages from unknown topic: {topic}");
+        MessageAcceptance::Ignore
+    };
+    Some(Verdict {
+        message_id,
+        source,
+        acceptance,
+    })
 }
 
 /// Saturating: these timestamps only feed peer latency estimates, so a skewed clock should degrade
@@ -934,7 +956,19 @@ async fn handle_forest_behaviour_event(
             .await
         }
         ForestBehaviourEvent::Gossipsub(e) => {
-            handle_gossip_event(e, network_sender_out, pubsub_block_str, pubsub_msg_str).await
+            if let Some(Verdict {
+                message_id,
+                source,
+                acceptance,
+            }) =
+                handle_gossip_event(e, network_sender_out, pubsub_block_str, pubsub_msg_str).await
+            {
+                swarm.behaviour_mut().report_message_validation_result(
+                    &message_id,
+                    &source,
+                    acceptance,
+                );
+            }
         }
         ForestBehaviourEvent::Hello(rr_event) => {
             let behaviour_mut = swarm.behaviour_mut();

src/libp2p/tests/gossip_validation_test.rs

@@ -0,0 +1,127 @@
+// Copyright 2019-2026 ChainSafe Systems
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use libp2p::{
+    PeerId,
+    gossipsub::{self, MessageAcceptance, MessageId, TopicHash},
+};
+
+use crate::{
+    blocks::{CachingBlockHeader, GossipBlock, RawBlockHeader},
+    libp2p::{
+        NetworkEvent, PubsubMessage, PubsubTopic,
+        service::{Verdict, handle_gossip_event},
+    },
+    message::SignedMessage,
+    shim::{
+        address::Address,
+        crypto::Signature,
+        message::{Message, Message_v3},
+    },
+    test_utils::dummy_ticket,
+};
+
+const NETWORK: &str = "testgossipvalidation";
+
+// `MessageAcceptance` doesn't implement `PartialEq`.
+fn assert_verdict(actual: &MessageAcceptance, expected: MessageAcceptance) {
+    assert_eq!(
+        std::mem::discriminant(actual),
+        std::mem::discriminant(&expected),
+        "verdict {actual:?}, expected {expected:?}"
+    );
+}
+
+fn gossip_message_event(data: Vec<u8>, topic: TopicHash) -> gossipsub::Event {
+    gossipsub::Event::Message {
+        propagation_source: PeerId::random(),
+        message_id: MessageId::new(b"test"),
+        message: gossipsub::Message {
+            source: None,
+            data,
+            sequence_number: None,
+            topic,
+        },
+    }
+}
+
+fn gossip_block() -> Vec<u8> {
+    fvm_ipld_encoding::to_vec(&GossipBlock {
+        header: CachingBlockHeader::new(RawBlockHeader {
+            miner_address: Address::new_id(0),
+            ticket: dummy_ticket(0),
+            ..Default::default()
+        }),
+        bls_messages: vec![],
+        secpk_messages: vec![],
+    })
+    .unwrap()
+}
+
+fn gossip_message() -> Vec<u8> {
+    let message: Message = Message_v3 {
+        to: Address::new_id(0).into(),
+        from: Address::new_id(0).into(),
+        ..Message_v3::default()
+    }
+    .into();
+    fvm_ipld_encoding::to_vec(&SignedMessage::new_unchecked(
+        message,
+        Signature::new_secp256k1(vec![0]),
+    ))
+    .unwrap()
+}
+
+/// Blocks and messages are accepted once decoded; anything else is ignored.
+#[rstest::rstest]
+#[case::block(
+    Some(PubsubTopic::Blocks),
+    gossip_block(),
+    MessageAcceptance::Accept,
+    true
+)]
+#[case::message(
+    Some(PubsubTopic::Messages),
+    gossip_message(),
+    MessageAcceptance::Accept,
+    true
+)]
+#[case::undecodable_block(Some(PubsubTopic::Blocks), vec![0xff], MessageAcceptance::Ignore, false)]
+#[case::undecodable_message(Some(PubsubTopic::Messages), vec![0xff], MessageAcceptance::Ignore, false)]
+#[case::unknown_topic(None, gossip_block(), MessageAcceptance::Ignore, false)]
+#[tokio::test]
+async fn filecoin_gossip_verdicts(
+    #[case] kind: Option<PubsubTopic>,
+    #[case] data: Vec<u8>,
+    #[case] expected: MessageAcceptance,
+    #[case] emitted: bool,
+) {
+    let block_topic = PubsubTopic::Blocks.ident(NETWORK).to_string();
+    let msg_topic = PubsubTopic::Messages.ident(NETWORK).to_string();
+    let topic = TopicHash::from_raw(match kind {
+        Some(kind) => kind.ident(NETWORK).to_string(),
+        None => "/unknown/topic".into(),
+    });
+    let (tx, rx) = flume::unbounded();
+
+    let Verdict {
+        acceptance: verdict,
+        ..
+    } = handle_gossip_event(
+        gossip_message_event(data, topic),
+        &tx,
+        &block_topic,
+        &msg_topic,
+    )
+    .await
+    .expect("an immediate verdict");
+
+    assert_verdict(&verdict, expected);
+    match rx.try_recv() {
+        Ok(NetworkEvent::PubsubMessage {
+            message: PubsubMessage::Block(_) | PubsubMessage::Message(_),
+        }) => assert!(emitted, "nothing should be emitted"),
+        Ok(other) => panic!("unexpected event: {other:?}"),
+        Err(_) => assert!(!emitted, "a decoded message should be emitted"),
+    }
+}