create validation stub to facilitate parallel work on validation and message passing

anchor/network/src/lib.rs

@@ -8,6 +8,8 @@ mod keypair_utils;
 mod network;
 mod transport;
 pub mod types;
+mod validation;
+
 pub use config::Config;
 pub use lighthouse_network::{ListenAddr, ListenAddress};
 pub use network::Network;

anchor/network/src/network.rs

@@ -1,3 +1,4 @@
+use crate::network::gossipsub::MessageId;
 use std::num::{NonZeroU8, NonZeroUsize};
 use std::pin::Pin;
 use std::sync::Arc;
@@ -6,32 +7,31 @@ use std::time::Duration;
 use futures::StreamExt;
 use libp2p::core::muxing::StreamMuxerBox;
 use libp2p::core::transport::Boxed;
-use libp2p::gossipsub::{IdentTopic, MessageAuthenticity, ValidationMode};
+use libp2p::gossipsub::{
+    IdentTopic, Message, MessageAcceptance, MessageAuthenticity, ValidationMode,
+};
 use libp2p::identity::Keypair;
 use libp2p::multiaddr::Protocol;
 use libp2p::swarm::SwarmEvent;
 use libp2p::{futures, gossipsub, identify, ping, PeerId, Swarm, SwarmBuilder};
 use lighthouse_network::discovery::DiscoveredPeers;
 use lighthouse_network::discv5::enr::k256::sha2::{Digest, Sha256};
+use lighthouse_network::EnrExt;
+use parking_lot::RwLock;
+use subnet_tracker::{SubnetEvent, SubnetId};
 use task_executor::TaskExecutor;
+use tokio::sync::mpsc;
 use tracing::{debug, error, info, trace, warn};
 
 use crate::behaviour::AnchorBehaviour;
 use crate::behaviour::AnchorBehaviourEvent;
 use crate::discovery::{Discovery, FIND_NODE_QUERY_CLOSEST_PEERS};
+use crate::handshake::node_info::{NodeInfo, NodeMetadata};
+use crate::handshake::{Behaviour, Event};
 use crate::keypair_utils::load_private_key;
 use crate::transport::build_transport;
 use crate::Config;
 
-use crate::handshake::node_info::{NodeInfo, NodeMetadata};
-use crate::handshake::{Behaviour, Event};
-use crate::types::ssv_message::SignedSSVMessage;
-use lighthouse_network::EnrExt;
-use parking_lot::RwLock;
-use ssz::Decode;
-use subnet_tracker::{SubnetEvent, SubnetId};
-use tokio::sync::mpsc;
-
 pub struct Network {
     swarm: Swarm<AnchorBehaviour>,
     subnet_event_receiver: mpsc::Receiver<SubnetEvent>,
@@ -120,26 +120,17 @@ impl Network {
                                         message_id,
                                         message,
                                     } => {
-                                        debug!(
-                                            source = ?propagation_source,
-                                            id = ?message_id,
-                                            "Received SignedSSVMessage"
+                                        self.on_message_received(
+                                            propagation_source,
+                                            &message_id,
+                                            message,
                                         );
-                                        match SignedSSVMessage::from_ssz_bytes(&message.data) {
-                                            Ok(deserialized_message) => {
-                                                debug!(msg = ?deserialized_message, "SignedSSVMessage deserialized");
-                                            }
-                                            Err(e) => {
-                                                error!("error" = ?e, "Failed to deserialize SignedSSVMessage");
-                                            }
-                                        }
                                     }
                                     // TODO handle gossipsub events
                                     _ => {
                                         debug!(event = ?ge, "Unhandled gossipsub event");
                                     }
                                 }
-                                // TODO handle gossipsub events
                             },
                             // Inform the peer manager about discovered peers.
                             //
@@ -185,6 +176,37 @@ impl Network {
         }
     }
 
+    fn on_message_received(
+        &mut self,
+        propagation_source: PeerId,
+        message_id: &MessageId,
+        message: Message,
+    ) {
+        debug!(
+            source = ?propagation_source,
+            id = ?message_id,
+            "Received SignedSSVMessage"
+        );
+
+        let result = self.validate_message(message);
+        let acceptance = if let Err(f) = &result {
+            f.into()
+        } else {
+            MessageAcceptance::Accept
+        };
+        let _ = self
+            .swarm
+            .behaviour_mut()
+            .gossipsub
+            .report_message_validation_result(message_id, &propagation_source, acceptance);
+
+        let Ok(_message) = result else {
+            return;
+        };
+
+        // todo pass on to app
+    }
+
     fn on_subnet_tracker_event(&mut self, event: SubnetEvent) {
         match event {
             SubnetEvent::Join(subnet) => {

anchor/network/src/validation.rs

@@ -0,0 +1,116 @@
+use crate::types::ssv_message::SignedSSVMessage;
+use crate::Network;
+use libp2p::gossipsub::{Message, MessageAcceptance};
+use ssz::Decode;
+use tracing::debug;
+
+// TODO taken from go-SSV as rough guidance. feel free to adjust as needed. https://github.com/ssvlabs/ssv/blob/e12abf7dfbbd068b99612fa2ebbe7e3372e57280/message/validation/errors.go#L55
+pub enum ValidationFailure {
+    WrongDomain,
+    NoShareMetadata,
+    UnknownValidator,
+    ValidatorLiquidated,
+    ValidatorNotAttesting,
+    EarlySlotMessage,
+    LateSlotMessage,
+    SlotAlreadyAdvanced,
+    RoundAlreadyAdvanced,
+    DecidedWithSameSigners,
+    PubSubDataTooBig(usize),
+    IncorrectTopic,
+    NonExistentCommitteeID,
+    RoundTooHigh,
+    ValidatorIndexMismatch,
+    TooManyDutiesPerEpoch,
+    NoDuty,
+    EstimatedRoundNotInAllowedSpread,
+    EmptyData,
+    MismatchedIdentifier,
+    SignatureVerification,
+    PubSubMessageHasNoData,
+    MalformedPubSubMessage(ssz::DecodeError),
+    NilSignedSSVMessage,
+    NilSSVMessage,
+    SSVDataTooBig,
+    InvalidRole,
+    UnexpectedConsensusMessage,
+    NoSigners,
+    WrongRSASignatureSize,
+    ZeroSigner,
+    SignerNotInCommittee,
+    DuplicatedSigner,
+    SignerNotLeader,
+    SignersNotSorted,
+    InconsistentSigners,
+    InvalidHash,
+    FullDataHash,
+    UndecodableMessageData,
+    EventMessage,
+    UnknownSSVMessageType,
+    UnknownQBFTMessageType,
+    InvalidPartialSignatureType,
+    PartialSignatureTypeRoleMismatch,
+    NonDecidedWithMultipleSigners,
+    DecidedNotEnoughSigners,
+    DifferentProposalData,
+    MalformedPrepareJustifications,
+    UnexpectedPrepareJustifications,
+    MalformedRoundChangeJustifications,
+    UnexpectedRoundChangeJustifications,
+    NoPartialSignatureMessages,
+    NoValidators,
+    NoSignatures,
+    SignersAndSignaturesWithDifferentLength,
+    PartialSigOneSigner,
+    PrepareOrCommitWithFullData,
+    FullDataNotInConsensusMessage,
+    TripleValidatorIndexInPartialSignatures,
+    ZeroRound,
+    DuplicatedMessage,
+    InvalidPartialSignatureTypeCount,
+    TooManyPartialSignatureMessages,
+    EncodeOperators,
+}
+
+impl From<&ValidationFailure> for MessageAcceptance {
+    fn from(value: &ValidationFailure) -> Self {
+        match value {
+            ValidationFailure::WrongDomain
+            | ValidationFailure::NoShareMetadata
+            | ValidationFailure::UnknownValidator
+            | ValidationFailure::ValidatorLiquidated
+            | ValidationFailure::ValidatorNotAttesting
+            | ValidationFailure::EarlySlotMessage
+            | ValidationFailure::LateSlotMessage
+            | ValidationFailure::SlotAlreadyAdvanced
+            | ValidationFailure::RoundAlreadyAdvanced
+            | ValidationFailure::DecidedWithSameSigners
+            | ValidationFailure::PubSubDataTooBig(_)
+            | ValidationFailure::IncorrectTopic
+            | ValidationFailure::NonExistentCommitteeID
+            | ValidationFailure::RoundTooHigh
+            | ValidationFailure::ValidatorIndexMismatch
+            | ValidationFailure::TooManyDutiesPerEpoch
+            | ValidationFailure::NoDuty
+            | ValidationFailure::EstimatedRoundNotInAllowedSpread => MessageAcceptance::Ignore,
+            _ => MessageAcceptance::Reject,
+        }
+    }
+}
+
+impl Network {
+    pub(crate) fn validate_message(
+        &mut self,
+        message: Message,
+    ) -> Result<SignedSSVMessage, ValidationFailure> {
+        // todo pre-deserialization validation
+
+        let message = SignedSSVMessage::from_ssz_bytes(&message.data)
+            .map_err(ValidationFailure::MalformedPubSubMessage)?;
+        debug!(msg = ?message, "SignedSSVMessage deserialized");
+
+        // todo post-deserialization validation
+
+        Ok(message)
+    }
+}