From 279d4c2c8013b3da0936fe7e045043562a2f26a8 Mon Sep 17 00:00:00 2001 From: shane-moore Date: Wed, 29 Jul 2026 19:15:09 -0700 Subject: [PATCH 1/2] fix: validate outbound messages statelessly --- anchor/client/src/lib.rs | 1 - anchor/message_sender/src/network.rs | 51 +++------ anchor/message_validator/src/lib.rs | 164 ++++++++++++++++++++++++--- 3 files changed, 164 insertions(+), 52 deletions(-) diff --git a/anchor/client/src/lib.rs b/anchor/client/src/lib.rs index d100a64b3..4c69e070a 100644 --- a/anchor/client/src/lib.rs +++ b/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(), }, diff --git a/anchor/message_sender/src/network.rs b/anchor/message_sender/src/network.rs index 6b69f030a..4fa7c927f 100644 --- a/anchor/message_sender/src/network.rs +++ b/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; use openssl::{ hash::MessageDigest, pkey::{PKey, Private}, @@ -15,7 +15,7 @@ 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}; @@ -23,30 +23,28 @@ const SIGNER_NAME: &str = "message_sign_and_send"; const SENDER_NAME: &str = "message_send"; /// Configuration for creating a NetworkMessageSender -pub struct NetworkMessageSenderConfig { +pub struct NetworkMessageSenderConfig { 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)>, pub private_key: Rsa, pub operator_id: OwnOperatorId, - pub validator: Option>>, pub is_synced: watch::Receiver, pub subnet_service: Arc>, } -pub struct NetworkMessageSender { +pub struct NetworkMessageSender { processor: processor::Senders, /// Channel to send messages to the network. Tuple of (topic string, message bytes). network_tx: mpsc::Sender<(String, Vec)>, private_key: PKey, operator_id: OwnOperatorId, - validator: Option>>, is_synced: watch::Receiver, subnet_service: Arc>, } -impl MessageSender for Arc> { +impl MessageSender for Arc> { fn sign_and_send( &self, message: UnsignedSSVMessage, @@ -118,8 +116,8 @@ impl MessageSender for Arc NetworkMessageSender { - pub fn new(config: NetworkMessageSenderConfig) -> Result, String> { +impl NetworkMessageSender { + pub fn new(config: NetworkMessageSenderConfig) -> Result, 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 { @@ -127,7 +125,6 @@ impl NetworkMessageSender { 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, })) @@ -136,32 +133,16 @@ impl NetworkMessageSender { 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_bytes) { + 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; } diff --git a/anchor/message_validator/src/lib.rs b/anchor/message_validator/src/lib.rs index 2c8fc12c5..48ee0d3be 100644 --- a/anchor/message_validator/src/lib.rs +++ b/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,35 @@ pub struct Validator { spec: Arc, } +/// 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_data: &[u8]) -> Result { + let signed_ssv_message = SignedSSVMessage::from_ssz_bytes(message_data) + .map_err(ValidationFailure::UndecodableMessageData)?; + 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 { + signed_ssv_message + .validate() + .map_err(ValidationFailure::from)?; + signed_ssv_message + .ssv_message() + .msg_id() + .role() + .ok_or(ValidationFailure::InvalidRole) +} + impl Validator { #[expect(clippy::too_many_arguments)] pub fn new( @@ -447,17 +474,8 @@ impl Validator { signed_ssv_message: &SignedSSVMessage, topic_context: &TopicContext, ) -> Result { - // 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 +1230,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 +1245,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, + ) -> 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. From 469598c39637674cadf4539a6a9f78fb15e8fee5 Mon Sep 17 00:00:00 2001 From: shane-moore Date: Wed, 29 Jul 2026 19:39:27 -0700 Subject: [PATCH 2/2] perf: avoid re-decoding outbound messages --- anchor/message_sender/src/network.rs | 7 +++---- anchor/message_validator/src/lib.rs | 15 +++++++++++---- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/anchor/message_sender/src/network.rs b/anchor/message_sender/src/network.rs index 4fa7c927f..0f23303dd 100644 --- a/anchor/message_sender/src/network.rs +++ b/anchor/message_sender/src/network.rs @@ -1,7 +1,7 @@ use std::sync::Arc; use database::OwnOperatorId; -use message_validator::validate_outbound; +use message_validator::validate_outbound_message; use openssl::{ hash::MessageDigest, pkey::{PKey, Private}, @@ -131,9 +131,7 @@ impl NetworkMessageSender { } fn do_send(&self, message: SignedSSVMessage, committee_id: CommitteeId) { - let message_bytes = message.as_ssz_bytes(); - - let message_slot = match validate_outbound(&message_bytes) { + let message_slot = match validate_outbound_message(&message) { Ok(slot) => slot, Err(err) => { error!( @@ -147,6 +145,7 @@ impl NetworkMessageSender { return; } }; + let message_bytes = message.as_ssz_bytes(); // Use subnet service for slot-based subnet calculation let subnet = match self diff --git a/anchor/message_validator/src/lib.rs b/anchor/message_validator/src/lib.rs index 48ee0d3be..15224e196 100644 --- a/anchor/message_validator/src/lib.rs +++ b/anchor/message_validator/src/lib.rs @@ -386,16 +386,23 @@ pub struct Validator { spec: Arc, } +/// Decode and perform stateless structural validation of an outbound message. +pub fn validate_outbound(message_data: &[u8]) -> Result { + 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_data: &[u8]) -> Result { - let signed_ssv_message = SignedSSVMessage::from_ssz_bytes(message_data) - .map_err(ValidationFailure::UndecodableMessageData)?; - validate_structure_and_role(&signed_ssv_message)?; +pub fn validate_outbound_message( + signed_ssv_message: &SignedSSVMessage, +) -> Result { + validate_structure_and_role(signed_ssv_message)?; signed_ssv_message .ssv_message() .extract_slot()