feat(rpc): validate external message accept before broadcast

core/src/blockchain_rpc/broadcast_listener.rs

@@ -6,6 +6,7 @@ use bytes::Bytes;
 use futures_util::FutureExt;
 use futures_util::future::BoxFuture;
 use tycho_network::InboundRequestMeta;
+use tycho_types::prelude::Cell;
 
 use super::SelfBroadcastListener;
 
@@ -19,6 +20,13 @@ pub trait BroadcastListener: Send + Sync + 'static {
     ) -> Self::HandleMessageFut<'_>;
 }
 
+/// Optional validator for external messages received via overlay broadcast.
+#[async_trait::async_trait]
+pub trait ExternalMessageValidator: Send + Sync + 'static {
+    /// Validate an external message cell.
+    async fn validate(&self, msg_cell: Cell) -> anyhow::Result<()>;
+}
+
 macro_rules! impl_listener_tuple {
     ($join_fn:path, {
         $($n:tt: $var:ident = $ty:ident,)*

core/src/blockchain_rpc/mod.rs

@@ -1,5 +1,6 @@
 pub use self::broadcast_listener::{
-    BoxBroadcastListener, BroadcastListener, BroadcastListenerExt, NoopBroadcastListener,
+    BoxBroadcastListener, BroadcastListener, BroadcastListenerExt, ExternalMessageValidator,
+    NoopBroadcastListener,
 };
 pub use self::client::{
     BlockDataFull, BlockDataFullWithNeighbour, BlockchainRpcClient, BlockchainRpcClientBuilder,
@@ -14,6 +15,7 @@ pub use self::rate_limits::{BlockchainRpcRateLimitsConfig, BlockchainRpcTrafficL
 pub use self::service::S3ProxyConfig;
 pub use self::service::{
     BlockchainRpcService, BlockchainRpcServiceBuilder, BlockchainRpcServiceConfig,
+    ExternalMessageValidatorHandle,
 };
 
 mod broadcast_listener;

core/src/blockchain_rpc/service.rs

@@ -1,18 +1,20 @@
 use std::num::NonZeroU32;
-use std::sync::Arc;
+use std::sync::{Arc, OnceLock};
 
 use anyhow::Context;
 use bytes::{Buf, Bytes};
 #[cfg(feature = "s3")]
 use bytesize::ByteSize;
 use serde::{Deserialize, Serialize};
-use tycho_block_util::message::validate_external_message;
+use tycho_block_util::message::parse_external_message;
 use tycho_network::{Response, Service, ServiceRequest, try_handle_prefix};
 use tycho_types::models::BlockId;
 use tycho_util::futures::BoxFutureOrNoop;
 use tycho_util::metrics::HistogramGuard;
 
-use crate::blockchain_rpc::broadcast_listener::{BroadcastListener, NoopBroadcastListener};
+use crate::blockchain_rpc::broadcast_listener::{
+    BroadcastListener, ExternalMessageValidator, NoopBroadcastListener,
+};
 use crate::blockchain_rpc::providers::{IntoRpcDataProvider, RpcDataProvider};
 use crate::blockchain_rpc::rate_limits::BlockchainRpcRateLimitsConfig;
 use crate::blockchain_rpc::{BAD_REQUEST_ERROR_CODE, INTERNAL_ERROR_CODE, NOT_FOUND_ERROR_CODE};
@@ -94,6 +96,8 @@ pub struct BlockchainRpcServiceBuilder<MandatoryFields> {
     mandatory_fields: MandatoryFields,
 }
 
+pub type ExternalMessageValidatorHandle = Arc<OnceLock<Arc<dyn ExternalMessageValidator>>>;
+
 impl<B> BlockchainRpcServiceBuilder<(B, CoreStorage, Arc<dyn RpcDataProvider>)>
 where
     B: BroadcastListener,
@@ -107,6 +111,7 @@ where
                 rpc_data_provider,
                 config: self.config,
                 broadcast_listener,
+                external_message_validator: Default::default(),
             }),
         }
     }
@@ -126,6 +131,7 @@ where
                 rpc_data_provider,
                 config: self.config,
                 broadcast_listener,
+                external_message_validator: Default::default(),
             }),
         }
     }
@@ -208,6 +214,12 @@ impl<B> Clone for BlockchainRpcService<B> {
     }
 }
 
+impl<B> BlockchainRpcService<B> {
+    pub fn external_message_validator_handle(&self) -> ExternalMessageValidatorHandle {
+        Arc::clone(&self.inner.external_message_validator)
+    }
+}
+
 impl BlockchainRpcService<()> {
     pub fn builder() -> BlockchainRpcServiceBuilder<((), (), ())> {
         BlockchainRpcServiceBuilder {
@@ -406,8 +418,18 @@ impl<B: BroadcastListener> Service<ServiceRequest> for BlockchainRpcService<B> {
                 metrics::counter!("tycho_rpc_broadcast_external_message_rx_bytes_total")
                     .increment(req.body.len() as u64);
                 BoxFutureOrNoop::future(async move {
-                    if let Err(e) = validate_external_message(&req.body).await {
-                        tracing::debug!("invalid external message: {e:?}");
+                    let msg_cell = match parse_external_message(&req.body).await {
+                        Ok(cell) => cell,
+                        Err(e) => {
+                            tracing::debug!("invalid external message: {e:?}");
+                            return;
+                        }
+                    };
+
+                    if let Some(validator) = inner.external_message_validator.get()
+                        && let Err(e) = validator.validate(msg_cell).await
+                    {
+                        tracing::debug!("external message rejected by validator: {e:?}");
                         return;
                     }
 
@@ -430,6 +452,7 @@ struct Inner<B> {
     config: BlockchainRpcServiceConfig,
     broadcast_listener: B,
     rpc_data_provider: Arc<dyn RpcDataProvider>,
+    external_message_validator: ExternalMessageValidatorHandle,
 }
 
 impl<B> Inner<B> {

core/src/node/mod.rs

@@ -22,8 +22,9 @@ use crate::block_strider::{
 #[cfg(feature = "s3")]
 use crate::blockchain_rpc::S3RpcDataProvider;
 use crate::blockchain_rpc::{
-    BlockchainRpcClient, BlockchainRpcService, BroadcastListener, IntoRpcDataProvider,
-    SelfBroadcastListener, StorageRpcDataProvider,
+    BlockchainRpcClient, BlockchainRpcService, BroadcastListener, ExternalMessageValidator,
+    ExternalMessageValidatorHandle, IntoRpcDataProvider, SelfBroadcastListener,
+    StorageRpcDataProvider,
 };
 use crate::global_config::{GlobalConfig, ZerostateId};
 use crate::overlay_client::{PublicOverlayClient, ValidatorsResolver};
@@ -74,6 +75,7 @@ pub struct NodeBase {
     pub core_storage: CoreStorage,
 
     pub blockchain_rpc_client: BlockchainRpcClient,
+    pub external_message_validator_handle: ExternalMessageValidatorHandle,
 
     #[cfg(feature = "s3")]
     pub s3_client: Option<S3Client>,
@@ -233,6 +235,14 @@ impl NodeBase {
         StorageBlockProvider::new(self.core_storage.clone())
     }
 
+    pub fn set_external_message_validator(&self, validator: Arc<dyn ExternalMessageValidator>) {
+        self.external_message_validator_handle
+            .set(validator)
+            .unwrap_or_else(|_| {
+                panic!("external message validator already initialized");
+            });
+    }
+
     /// Creates a new [`BlockStrider`] using options from the base config.
     pub fn build_strider<P, S>(
         &self,
@@ -331,33 +341,38 @@ impl<'a> NodeBaseBuilder<'a, init::Step1> {
             .transpose()
             .context("failed to create S3 client")?;
 
-        let (_, blockchain_rpc_client) = self.step.prev_step.net.add_blockchain_rpc(
-            &self.common.global_config.zerostate,
-            self.step.store.core_storage.clone(),
-            (
-                StorageRpcDataProvider::new(self.step.store.core_storage.clone()),
-                #[cfg(feature = "s3")]
-                s3_client.clone().and_then(|s3_client| {
-                    let proxy_config = self
-                        .common
-                        .base_config
-                        .blockchain_rpc_service
-                        .s3_proxy
-                        .as_ref()?; // when s3_proxy = None - ignore rpc provider
-                    let storage = self.step.store.core_storage.clone();
-                    Some(S3RpcDataProvider::new(s3_client, storage, proxy_config))
-                }),
-            ),
-            remote_broadcast_listener,
-            self_broadcast_listener,
-            self.common.base_config,
-        );
+        let (blockchain_rpc_service, blockchain_rpc_client) =
+            self.step.prev_step.net.add_blockchain_rpc(
+                &self.common.global_config.zerostate,
+                self.step.store.core_storage.clone(),
+                (
+                    StorageRpcDataProvider::new(self.step.store.core_storage.clone()),
+                    #[cfg(feature = "s3")]
+                    s3_client.clone().and_then(|s3_client| {
+                        let proxy_config = self
+                            .common
+                            .base_config
+                            .blockchain_rpc_service
+                            .s3_proxy
+                            .as_ref()?; // when s3_proxy = None - ignore rpc provider
+                        let storage = self.step.store.core_storage.clone();
+                        Some(S3RpcDataProvider::new(s3_client, storage, proxy_config))
+                    }),
+                ),
+                remote_broadcast_listener,
+                self_broadcast_listener,
+                self.common.base_config,
+            );
+
+        let external_message_validator_handle =
+            blockchain_rpc_service.external_message_validator_handle();
 
         Ok(NodeBaseBuilder {
             common: self.common,
             step: init::Step2 {
                 prev_step: self.step,
                 blockchain_rpc_client,
+                external_message_validator_handle,
                 #[cfg(feature = "s3")]
                 s3_client,
             },
@@ -370,6 +385,7 @@ impl<'a> NodeBaseBuilder<'a, init::Final> {
         let net = self.step.prev_step.prev_step.net;
         let store = self.step.prev_step.store;
         let blockchain_rpc_client = self.step.blockchain_rpc_client;
+        let external_message_validator_handle = self.step.external_message_validator_handle;
         #[cfg(feature = "s3")]
         let s3_client = self.step.s3_client;
 
@@ -385,6 +401,7 @@ impl<'a> NodeBaseBuilder<'a, init::Final> {
             storage_context: store.context,
             core_storage: store.core_storage,
             blockchain_rpc_client,
+            external_message_validator_handle,
             #[cfg(feature = "s3")]
             s3_client,
         })
@@ -495,6 +512,7 @@ pub mod init {
     pub struct Step2 {
         pub(super) prev_step: Step1,
         pub(super) blockchain_rpc_client: BlockchainRpcClient,
+        pub(super) external_message_validator_handle: ExternalMessageValidatorHandle,
         #[cfg(feature = "s3")]
         pub(super) s3_client: Option<S3Client>,
     }

rpc/Cargo.toml

@@ -16,6 +16,7 @@ arc-swap = { workspace = true }
 axum = { workspace = true }
 axum-client-ip = { workspace = true }
 base64 = { workspace = true }
+async-trait = { workspace = true }
 bitflags = { workspace = true }
 bytes = { workspace = true }
 futures-util = { workspace = true }
@@ -41,6 +42,7 @@ weedb = { workspace = true }
 # local deps
 tycho-block-util = { workspace = true }
 tycho-core = { workspace = true }
+tycho-executor = { workspace = true }
 tycho-storage = { workspace = true }
 tycho-util = { workspace = true }
 tycho-rpc-subscriptions = { workspace = true }

rpc/src/config.rs

@@ -44,6 +44,11 @@ pub struct RpcConfig {
     /// Configuration of getter requests.
     pub run_get_method: RunGetMethodConfig,
 
+    /// Whether to validate external messages before broadcasting.
+    ///
+    /// Default: `false`.
+    pub validate_external_messages: bool,
+
     /// Subscriptions limits and buffering.
     pub subscriptions: SubscriptionsConfig,
 
@@ -157,6 +162,7 @@ impl Default for RpcConfig {
             allow_huge_requests: false,
             max_parallel_block_downloads: 10,
             run_get_method: RunGetMethodConfig::default(),
+            validate_external_messages: false,
             subscriptions: SubscriptionsConfig::default(),
             real_ip_source: ClientIpSource::ConnectInfo,
             rate_limits: None,

rpc/src/endpoint/jrpc/mod.rs

@@ -8,7 +8,7 @@ use axum::response::{IntoResponse, Response};
 use base64::prelude::{BASE64_STANDARD, Engine as _};
 use serde::{Deserialize, Serialize};
 use serde_json::value::RawValue;
-use tycho_block_util::message::{ExtMsgRepr, validate_external_message};
+use tycho_block_util::message::{ExtMsgRepr, parse_external_message};
 use tycho_types::models::*;
 use tycho_types::prelude::*;
 use tycho_util::metrics::HistogramGuard;
@@ -84,14 +84,23 @@ pub async fn route(State(state): State<RpcState>, req: Jrpc<RfcBehaviour, Method
             }
         }
         MethodParams::SendMessage(p) => {
-            if let Err(e) = validate_external_message(&p.message).await {
-                return JrpcErrorResponse {
-                    id: Some(req.id),
-                    code: INVALID_BOC_CODE,
-                    message: e.to_string().into(),
-                    behaviour: PhantomData::<RfcBehaviour>,
+            if state.config().validate_external_messages {
+                let msg_cell = match parse_external_message(&p.message).await {
+                    Ok(cell) => cell,
+                    Err(e) => {
+                        return JrpcErrorResponse {
+                            id: Some(req.id),
+                            code: INVALID_BOC_CODE,
+                            message: e.to_string().into(),
+                            behaviour: PhantomData::<RfcBehaviour>,
+                        }
+                        .into_response();
+                    }
+                };
+
+                if let Err(e) = state.check_external_message(msg_cell).await {
+                    return error_to_response(req.id, e);
                 }
-                .into_response();
             }
 
             state.broadcast_external_message(&p.message).await;
@@ -520,6 +529,10 @@ fn error_to_response(id: i64, e: RpcStateError) -> Response {
     let (code, message) = match e {
         RpcStateError::NotReady => (NOT_READY_CODE, Cow::Borrowed("not ready")),
         RpcStateError::NotSupported => (NOT_SUPPORTED_CODE, Cow::Borrowed("method not supported")),
+        RpcStateError::NotAccepted { exit_code } => (
+            INVALID_MESSAGE_CODE,
+            format!("message not accepted, exit code: {exit_code}").into(),
+        ),
         RpcStateError::Internal(e) => (INTERNAL_ERROR_CODE, e.to_string().into()),
         RpcStateError::BadRequest(e) => (INVALID_PARAMS_CODE, e.to_string().into()),
     };

rpc/src/endpoint/proto/mod.rs

@@ -5,7 +5,7 @@ use axum::extract::State;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use bytes::Bytes;
-use tycho_block_util::message::validate_external_message;
+use tycho_block_util::message::parse_external_message;
 use tycho_types::cell::HashBytes;
 use tycho_types::models::*;
 use tycho_types::prelude::*;
@@ -107,12 +107,21 @@ pub async fn route(State(state): State<RpcState>, Protobuf(req): Protobuf<Reques
             }
         }
         request::Call::SendMessage(p) => {
-            if let Err(e) = validate_external_message(&p.message).await {
-                return ProtoErrorResponse {
-                    code: INVALID_BOC_CODE,
-                    message: e.to_string().into(),
+            if state.config().validate_external_messages {
+                let msg_cell = match parse_external_message(&p.message).await {
+                    Ok(cell) => cell,
+                    Err(e) => {
+                        return ProtoErrorResponse {
+                            code: INVALID_BOC_CODE,
+                            message: e.to_string().into(),
+                        }
+                        .into_response();
+                    }
+                };
+
+                if let Err(e) = state.check_external_message(msg_cell).await {
+                    return error_to_response(e);
                 }
-                .into_response();
             }
 
             state.broadcast_external_message(&p.message).await;
@@ -436,6 +445,10 @@ fn error_to_response(e: RpcStateError) -> Response {
     let (code, message) = match e {
         RpcStateError::NotReady => (NOT_READY_CODE, Cow::Borrowed("not ready")),
         RpcStateError::NotSupported => (NOT_SUPPORTED_CODE, Cow::Borrowed("method not supported")),
+        RpcStateError::NotAccepted { exit_code } => (
+            INVALID_MESSAGE_CODE,
+            format!("message not accepted, exit code: {exit_code}").into(),
+        ),
         RpcStateError::Internal(e) => (INTERNAL_ERROR_CODE, e.to_string().into()),
         RpcStateError::BadRequest(e) => (INVALID_PARAMS_CODE, e.to_string().into()),
     };