fix(message_sender): validate outbound messages statelessly

anchor/client/src/lib.rs

@@ -562,7 +562,6 @@ impl Client {
                     network_tx: network_tx.clone(),
                     private_key: key.clone(),
                     operator_id: operator_id.clone(),
-                    validator: Some(message_validator.clone()),
                     is_synced: is_synced.clone(),
                     subnet_service: subnet_service.clone(),
                 },

anchor/message_sender/src/network.rs

@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use database::OwnOperatorId;
-use message_validator::{DutiesProvider, MessageAcceptance, TopicContext, Validator};
+use message_validator::validate_outbound_message;
 use openssl::{
     hash::MessageDigest,
     pkey::{PKey, Private},
@@ -15,38 +15,36 @@ use ssv_types::{
 use ssz::Encode;
 use subnet_service::SubnetService;
 use tokio::sync::{mpsc, mpsc::error::TrySendError, watch};
-use tracing::{debug, error, trace, warn};
+use tracing::{error, trace, warn};
 
 use crate::{Error, MessageCallback, MessageSender, SigningError};
 
 const SIGNER_NAME: &str = "message_sign_and_send";
 const SENDER_NAME: &str = "message_send";
 
 /// Configuration for creating a NetworkMessageSender
-pub struct NetworkMessageSenderConfig<S: SlotClock, D: DutiesProvider> {
+pub struct NetworkMessageSenderConfig<S: SlotClock> {
     pub processor: processor::Senders,
     /// Channel to send messages to the network. Tuple of (topic string, message bytes).
     /// Per SIP-43, the topic is determined by the message's slot.
     pub network_tx: mpsc::Sender<(String, Vec<u8>)>,
     pub private_key: Rsa<Private>,
     pub operator_id: OwnOperatorId,
-    pub validator: Option<Arc<Validator<S, D>>>,
     pub is_synced: watch::Receiver<bool>,
     pub subnet_service: Arc<SubnetService<S>>,
 }
 
-pub struct NetworkMessageSender<S: SlotClock, D: DutiesProvider> {
+pub struct NetworkMessageSender<S: SlotClock> {
     processor: processor::Senders,
     /// Channel to send messages to the network. Tuple of (topic string, message bytes).
     network_tx: mpsc::Sender<(String, Vec<u8>)>,
     private_key: PKey<Private>,
     operator_id: OwnOperatorId,
-    validator: Option<Arc<Validator<S, D>>>,
     is_synced: watch::Receiver<bool>,
     subnet_service: Arc<SubnetService<S>>,
 }
 
-impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMessageSender<S, D>> {
+impl<S: SlotClock + 'static> MessageSender for Arc<NetworkMessageSender<S>> {
     fn sign_and_send(
         &self,
         message: UnsignedSSVMessage,
@@ -118,54 +116,36 @@ impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMes
     }
 }
 
-impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageSender<S, D> {
-    pub fn new(config: NetworkMessageSenderConfig<S, D>) -> Result<Arc<Self>, String> {
+impl<S: SlotClock + 'static> NetworkMessageSender<S> {
+    pub fn new(config: NetworkMessageSenderConfig<S>) -> Result<Arc<Self>, String> {
         let private_key = PKey::from_rsa(config.private_key)
             .map_err(|err| format!("Failed to create PKey from RSA: {err}"))?;
         Ok(Arc::new(Self {
             processor: config.processor,
             network_tx: config.network_tx,
             private_key,
             operator_id: config.operator_id,
-            validator: config.validator,
             is_synced: config.is_synced,
             subnet_service: config.subnet_service,
         }))
     }
 
     fn do_send(&self, message: SignedSSVMessage, committee_id: CommitteeId) {
-        let message_bytes = message.as_ssz_bytes();
-
-        // For outgoing messages, we use default TopicContext (no topic validation)
-        // since we're just doing a sanity check on our own message content
-        if let Some(validator) = self.validator.as_ref()
-            && let Err(err) = validator
-                .validate(&message_bytes, &TopicContext::default())
-                .as_result()
-        {
-            // `Reject` is more severe and can be punished by other peers. We should not have
-            // created this message ever, while `Ignore` can be triggered simply because the message
-            // is irrelevant by now.
-            if let MessageAcceptance::Reject = MessageAcceptance::from(err) {
-                warn!(?err, "Validation of outgoing message failed (Reject)");
-                debug!(msg = %message, "Failing message");
-            } else {
-                debug!(?err, "Validation of outgoing message failed (Ignore)");
-            }
-            return;
-        }
-
-        // Extract slot from message for slot-based topic routing (per SIP-43)
-        let message_slot = match message.ssv_message().extract_slot() {
-            Some(slot) => slot,
-            None => {
-                warn!(
+        let message_slot = match validate_outbound_message(&message) {
+            Ok(slot) => slot,
+            Err(err) => {
+                error!(
+                    ?err,
                     ?committee_id,
-                    "Cannot extract slot from message for topic routing"
+                    ssv_msg_id = ?message.ssv_message().msg_id(),
+                    msg_type = ?message.ssv_message().msg_type(),
+                    role = ?message.ssv_message().msg_id().role(),
+                    "Stateless validation of outgoing message failed"
                 );
                 return;
             }
         };
+        let message_bytes = message.as_ssz_bytes();
 
         // Use subnet service for slot-based subnet calculation
         let subnet = match self

anchor/message_validator/src/lib.rs

@@ -329,7 +329,7 @@ impl ValidatedMessage {
 /// Context for topic-aware message validation.
 ///
 /// This enum makes explicit whether topic validation should be performed:
-/// - `SkipValidation`: Used for outgoing messages and tests where topic validation is not needed
+/// - `SkipValidation`: Used for tests where topic validation is not needed
 /// - `Validate`: Used for incoming network messages where topic validation is required
 ///
 /// For incoming network messages, always use `TopicContext::Validate`. If topic parsing
@@ -339,9 +339,7 @@ impl ValidatedMessage {
 pub enum TopicContext {
     /// Skip topic validation entirely.
     ///
-    /// Used for:
-    /// - Outgoing message self-validation (we calculate our own routing)
-    /// - Testing scenarios where topic context is irrelevant
+    /// Used for testing scenarios where topic context is irrelevant.
     #[default]
     SkipValidation,
 
@@ -388,6 +386,42 @@ pub struct Validator<S: SlotClock, D: DutiesProvider> {
     spec: Arc<ChainSpec>,
 }
 
+/// Decode and perform stateless structural validation of an outbound message.
+pub fn validate_outbound(message_data: &[u8]) -> Result<Slot, ValidationFailure> {
+    let signed_ssv_message = SignedSSVMessage::from_ssz_bytes(message_data)
+        .map_err(ValidationFailure::UndecodableMessageData)?;
+    validate_outbound_message(&signed_ssv_message)
+}
+
+/// Perform stateless structural validation of an outbound message and return its routing slot.
+///
+/// This is not an authorization boundary. It deliberately excludes all network, duty, timing,
+/// fork-role, signature-verification, and validation-state checks. Outbound producers must enforce
+/// those invariants before constructing the message. Incoming messages continue through
+/// [`Validator::validate`], which owns gossip validation state.
+pub fn validate_outbound_message(
+    signed_ssv_message: &SignedSSVMessage,
+) -> Result<Slot, ValidationFailure> {
+    validate_structure_and_role(signed_ssv_message)?;
+    signed_ssv_message
+        .ssv_message()
+        .extract_slot()
+        .ok_or(ValidationFailure::UnknownMessageSlot)
+}
+
+fn validate_structure_and_role(
+    signed_ssv_message: &SignedSSVMessage,
+) -> Result<Role, ValidationFailure> {
+    signed_ssv_message
+        .validate()
+        .map_err(ValidationFailure::from)?;
+    signed_ssv_message
+        .ssv_message()
+        .msg_id()
+        .role()
+        .ok_or(ValidationFailure::InvalidRole)
+}
+
 impl<S: SlotClock + 'static, D: DutiesProvider> Validator<S, D> {
     #[expect(clippy::too_many_arguments)]
     pub fn new(
@@ -447,17 +481,8 @@ impl<S: SlotClock + 'static, D: DutiesProvider> Validator<S, D> {
         signed_ssv_message: &SignedSSVMessage,
         topic_context: &TopicContext,
     ) -> Result<ValidatedMessage, ValidationFailure> {
-        // Structural validation: signer/signature invariants, RSA size, etc.
-        signed_ssv_message
-            .validate()
-            .map_err(ValidationFailure::from)?;
-
-        // Get the role from message ID
+        let role = validate_structure_and_role(signed_ssv_message)?;
         let ssv_message = signed_ssv_message.ssv_message();
-        let role = ssv_message
-            .msg_id()
-            .role()
-            .ok_or(ValidationFailure::InvalidRole)?;
 
         // Get committee ID for topic validation
         let committee_id = match ssv_message.msg_id().duty_executor() {
@@ -1212,7 +1237,7 @@ pub(crate) fn hash_data(full_data: &[u8]) -> [u8; 32] {
 mod tests {
     use std::{collections::HashMap, sync::Arc};
 
-    use bls::{Hash256, PublicKeyBytes};
+    use bls::{Hash256, PublicKeyBytes, Signature};
     use duties_tracker::{DutiesProvider, DutyAssignment};
     use openssl::{
         hash::MessageDigest,
@@ -1227,16 +1252,130 @@ mod tests {
         domain_type::DomainType,
         message::{MsgType, SSVMessage, SignedSSVMessage},
         msgid::{DutyExecutor, MessageId, Role},
+        partial_sig::{PartialSignatureKind, PartialSignatureMessage, PartialSignatureMessages},
     };
     use ssz::Encode;
     use types::{Epoch, Slot};
 
-    use crate::{MessageAcceptance, ValidationFailure, hash_data};
+    use crate::{MessageAcceptance, ValidationFailure, hash_data, validate_outbound};
 
     // Constants for committee sizes in tests to improve readability.
     pub(crate) const SINGLE_NODE_COMMITTEE: usize = 1;
     pub(crate) const FOUR_NODE_COMMITTEE: usize = 4;
 
+    fn signed_test_message(
+        msg_type: MsgType,
+        message_id: MessageId,
+        data: Vec<u8>,
+    ) -> SignedSSVMessage {
+        let ssv_message = SSVMessage::new(msg_type, message_id, data)
+            .expect("test SSVMessage should be structurally valid");
+        SignedSSVMessage::new(
+            vec![[0xAA; RSA_SIGNATURE_SIZE]],
+            vec![OperatorId(1)],
+            ssv_message,
+            vec![],
+        )
+        .expect("test SignedSSVMessage should be structurally valid")
+    }
+
+    #[test]
+    fn validate_outbound_returns_nested_message_slots() {
+        let consensus_message_id = create_message_id_for_test(Role::Committee);
+        let mut qbft_message = QbftMessageBuilder::new(Role::Committee, QbftMessageType::Proposal)
+            .with_identifier(consensus_message_id.clone())
+            .build();
+        qbft_message.height = 42;
+        let signed_consensus_message = signed_test_message(
+            MsgType::SSVConsensusMsgType,
+            consensus_message_id,
+            qbft_message.as_ssz_bytes(),
+        );
+
+        assert_eq!(
+            validate_outbound(&signed_consensus_message.as_ssz_bytes()),
+            Ok(Slot::new(42))
+        );
+
+        let partial_signature_messages = PartialSignatureMessages {
+            kind: PartialSignatureKind::RandaoPartialSig,
+            slot: Slot::new(43),
+            messages: VariableList::new(vec![PartialSignatureMessage {
+                partial_signature: Signature::empty(),
+                signing_root: Hash256::ZERO,
+                signer: OperatorId(1),
+                validator_index: ValidatorIndex(0),
+            }])
+            .expect("one partial signature should fit"),
+        };
+        let signed_partial_signature_message = signed_test_message(
+            MsgType::SSVPartialSignatureMsgType,
+            create_message_id_for_test(Role::Proposer),
+            partial_signature_messages.as_ssz_bytes(),
+        );
+
+        assert_eq!(
+            validate_outbound(&signed_partial_signature_message.as_ssz_bytes()),
+            Ok(Slot::new(43))
+        );
+    }
+
+    #[test]
+    fn validate_outbound_rejects_malformed_outer_and_nested_messages() {
+        assert!(matches!(
+            validate_outbound(&[]),
+            Err(ValidationFailure::UndecodableMessageData(_))
+        ));
+
+        for (msg_type, role) in [
+            (MsgType::SSVConsensusMsgType, Role::Committee),
+            (MsgType::SSVPartialSignatureMsgType, Role::Proposer),
+        ] {
+            let signed_message =
+                signed_test_message(msg_type, create_message_id_for_test(role), vec![0x01]);
+            assert_eq!(
+                validate_outbound(&signed_message.as_ssz_bytes()),
+                Err(ValidationFailure::UnknownMessageSlot)
+            );
+        }
+    }
+
+    #[test]
+    fn validate_outbound_rejects_duplicate_signers() {
+        let qbft_message =
+            QbftMessageBuilder::new(Role::Committee, QbftMessageType::Proposal).build();
+        let mut signed_message =
+            create_signed_consensus_message(qbft_message, vec![OperatorId(1)], vec![], vec![]);
+        signed_message
+            .aggregate([signed_message.clone()])
+            .expect("aggregation permits duplicate signers for validation tests");
+
+        assert_eq!(
+            validate_outbound(&signed_message.as_ssz_bytes()),
+            Err(ValidationFailure::DuplicatedSigner)
+        );
+    }
+
+    #[test]
+    fn validate_outbound_rejects_invalid_role() {
+        let mut invalid_message_id = [0u8; 56];
+        invalid_message_id[4] = u8::MAX;
+        let invalid_message_id = MessageId::from(invalid_message_id);
+        let qbft_message = QbftMessageBuilder::new(Role::Committee, QbftMessageType::Proposal)
+            .with_identifier(invalid_message_id.clone())
+            .build();
+        let signed_message = signed_test_message(
+            MsgType::SSVConsensusMsgType,
+            invalid_message_id,
+            qbft_message.as_ssz_bytes(),
+        );
+
+        assert_eq!(
+            validate_outbound(&signed_message.as_ssz_bytes()),
+            Err(ValidationFailure::InvalidRole)
+        );
+    }
+
     /// Test that an `ExcessiveDutyCount` maps to `Ignore`.
     /// Duty-limit breach is a rate condition. An honest relayer can forward a message that
     /// pushes a signer over its per-epoch duty count. Not a provable protocol violation.