diff --git a/Cargo.lock b/Cargo.lock index 606780fe9a..d906e134e5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4382,6 +4382,7 @@ dependencies = [ "ahash", "anyhow", "arc-swap", + "async-trait", "axum", "axum-client-ip", "base64 0.22.1", @@ -4405,6 +4406,7 @@ dependencies = [ "tracing", "tycho-block-util", "tycho-core", + "tycho-executor", "tycho-network", "tycho-rpc-subscriptions", "tycho-storage", diff --git a/core/src/blockchain_rpc/broadcast_listener.rs b/core/src/blockchain_rpc/broadcast_listener.rs index c818642b4f..cc70082e30 100644 --- a/core/src/blockchain_rpc/broadcast_listener.rs +++ b/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,)* diff --git a/core/src/blockchain_rpc/mod.rs b/core/src/blockchain_rpc/mod.rs index 1eb10e586f..e61cf59063 100644 --- a/core/src/blockchain_rpc/mod.rs +++ b/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; diff --git a/core/src/blockchain_rpc/service.rs b/core/src/blockchain_rpc/service.rs index 3dfd23628c..f042272853 100644 --- a/core/src/blockchain_rpc/service.rs +++ b/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 { mandatory_fields: MandatoryFields, } +pub type ExternalMessageValidatorHandle = Arc>>; + impl BlockchainRpcServiceBuilder<(B, CoreStorage, Arc)> 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 Clone for BlockchainRpcService { } } +impl BlockchainRpcService { + 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 Service for BlockchainRpcService { 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 { config: BlockchainRpcServiceConfig, broadcast_listener: B, rpc_data_provider: Arc, + external_message_validator: ExternalMessageValidatorHandle, } impl Inner { diff --git a/core/src/node/mod.rs b/core/src/node/mod.rs index 7dcd89746c..36a159a7a5 100644 --- a/core/src/node/mod.rs +++ 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, @@ -233,6 +235,14 @@ impl NodeBase { StorageBlockProvider::new(self.core_storage.clone()) } + pub fn set_external_message_validator(&self, validator: Arc) { + 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( &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, } diff --git a/rpc/Cargo.toml b/rpc/Cargo.toml index 73c36171c3..5c2eff6c75 100644 --- a/rpc/Cargo.toml +++ b/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 } diff --git a/rpc/res/config.boc b/rpc/res/config.boc new file mode 100644 index 0000000000..e8b83460f3 Binary files /dev/null and b/rpc/res/config.boc differ diff --git a/rpc/src/config.rs b/rpc/src/config.rs index 4f694a0620..3f63ac093b 100644 --- a/rpc/src/config.rs +++ b/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, diff --git a/rpc/src/endpoint/jrpc/mod.rs b/rpc/src/endpoint/jrpc/mod.rs index fc8e19b828..cd765193e9 100644 --- a/rpc/src/endpoint/jrpc/mod.rs +++ b/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, req: Jrpc { - 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::, + 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::, + } + .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()), }; diff --git a/rpc/src/endpoint/proto/mod.rs b/rpc/src/endpoint/proto/mod.rs index 208db2bab9..53930ab53f 100644 --- a/rpc/src/endpoint/proto/mod.rs +++ b/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, Protobuf(req): Protobuf { - 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()), }; diff --git a/rpc/src/node.rs b/rpc/src/node.rs index fb902a366e..3c5530e84a 100644 --- a/rpc/src/node.rs +++ b/rpc/src/node.rs @@ -1,3 +1,5 @@ +use std::sync::Arc; + use anyhow::{Context, Result}; use tycho_types::models::BlockId; @@ -53,6 +55,10 @@ impl NodeBaseInitRpc for tycho_core::node::NodeBase { rpc_state.init(last_block_id).await?; + if config.validate_external_messages { + self.set_external_message_validator(Arc::new(rpc_state.clone())); + } + let endpoint = rpc_state .bind_endpoint() .await diff --git a/rpc/src/state/mod.rs b/rpc/src/state/mod.rs index ff5d5971fb..7a455536d1 100644 --- a/rpc/src/state/mod.rs +++ b/rpc/src/state/mod.rs @@ -16,9 +16,10 @@ use tycho_block_util::state::{RefMcStateHandle, ShardStateStuff}; use tycho_core::block_strider::{ BlockSubscriber, BlockSubscriberContext, StateSubscriber, StateSubscriberContext, }; -use tycho_core::blockchain_rpc::BlockchainRpcClient; +use tycho_core::blockchain_rpc::{BlockchainRpcClient, ExternalMessageValidator}; use tycho_core::global_config::ZerostateId; use tycho_core::storage::{CoreStorage, KeyBlocksDirection}; +use tycho_executor::{Executor, ExecutorInspector, ExecutorParams, ParsedConfig, TxError}; use tycho_rpc_subscriptions::SubscriberManagerConfig; use tycho_types::models::*; use tycho_types::prelude::*; @@ -317,6 +318,109 @@ impl RpcState { .await; } + pub async fn check_external_message(&self, msg_cell: Cell) -> Result<(), RpcStateError> { + let bc_config = self.inner.blockchain_config.load_full(); + let parsed_config = bc_config + .parsed_config + .clone() + .ok_or(RpcStateError::NotReady)?; + + // Extract destination address from the message. + let dst = { + let mut slice = msg_cell.as_slice_allow_exotic(); + let Ok(MsgInfo::ExtIn(info)) = MsgInfo::load_from(&mut slice) else { + return Err(RpcStateError::bad_request(anyhow::anyhow!( + "invalid message" + ))); + }; + let IntAddr::Std(dst) = info.dst else { + return Err(RpcStateError::bad_request(anyhow::anyhow!( + "invalid message" + ))); + }; + if dst.anycast.is_some() { + return Err(RpcStateError::bad_request(anyhow::anyhow!( + "invalid message" + ))); + } + dst + }; + + // Get the current account state. + let LoadedAccountState::Found { + state: shard_account, + mc_ref_handle, + .. + } = self.inner.get_account_state(&dst)? + else { + return Ok(()); + }; + + let libraries = if dst.is_masterchain() { + self.inner + .mc_accounts + .read() + .as_ref() + .map(|c| c.libraries.clone()) + .unwrap_or_default() + } else { + let cache = self.inner.sc_accounts.read(); + cache + .iter() + .find(|(shard, _)| shard.contains_account(&dst.address)) + .map(|(_, c)| c.libraries.clone()) + .unwrap_or_default() + }; + + let mc_info = self.inner.mc_info.read().clone(); + let block_lt = mc_info.timings.gen_lt; + let prev_mc_block_id = *mc_info.block_id; + + tycho_util::sync::rayon_run(move || { + let _guard = mc_ref_handle; + + let global_id = parsed_config.global_id; + let capabilities = parsed_config.global.capabilities; + + let params = ExecutorParams { + libraries, + rand_seed: HashBytes::ZERO, + block_lt, + block_unixtime: now_sec(), + prev_mc_block_id: Some(prev_mc_block_id), + disable_delete_frozen_accounts: true, + full_body_in_bounced: capabilities.contains(GlobalCapability::CapFullBodyInBounced), + charge_action_fees_on_fail: true, + strict_extra_currency: true, + authority_marks_enabled: capabilities.contains(GlobalCapability::CapSuspendByMarks), + vm_modifiers: tycho_vm::BehaviourModifiers { + signature_with_id: capabilities + .contains(GlobalCapability::CapSignatureWithId) + .then_some(global_id), + enable_signature_domains: capabilities + .contains(GlobalCapability::CapSignatureDomain), + ..Default::default() + }, + }; + + let executor = Executor::new(¶ms, &parsed_config); + let mut inspector = ExecutorInspector::default(); + match executor.check_ordinary_ext(&dst, msg_cell, &shard_account, Some(&mut inspector)) + { + Ok(()) => Ok(()), + Err(TxError::Skipped | TxError::Fatal(_)) => match inspector.exit_code { + Some(exit_code) => Err(RpcStateError::NotAccepted { exit_code }), + None => Err(RpcStateError::bad_request(anyhow::anyhow!( + "invalid message" + ))), + }, + } + }) + .await + } +} + +impl RpcState { pub fn get_unpacked_blockchain_config(&self) -> Arc { self.inner.blockchain_config.load_full() } @@ -558,6 +662,15 @@ impl RpcState { } } +#[async_trait::async_trait] +impl ExternalMessageValidator for RpcState { + async fn validate(&self, msg_cell: Cell) -> Result<()> { + self.check_external_message(msg_cell) + .await + .map_err(|e| anyhow::anyhow!(e)) + } +} + pub struct RpcStateSubscriber { inner: Arc, } @@ -662,6 +775,7 @@ pub struct LatestBlockchainConfig { pub raw: BlockchainConfigParams, pub unpacked: tycho_vm::UnpackedConfig, pub modifiers: tycho_vm::BehaviourModifiers, + pub parsed_config: Option>, } impl Default for LatestBlockchainConfig { @@ -678,6 +792,7 @@ impl Default for LatestBlockchainConfig { size_limits_config: None, }, modifiers: Default::default(), + parsed_config: None, } } } @@ -788,7 +903,9 @@ impl Inner { } // Fill config. - if let Some(config) = load_blockchain_config(&mc_state) { + if let Some(config) = + load_blockchain_config(&mc_state, self.config.validate_external_messages) + { self.blockchain_config.store(config); } @@ -1046,7 +1163,9 @@ impl Inner { if shard.is_masterchain() { // Fill config. - if let Some(config) = load_blockchain_config(state) { + if let Some(config) = + load_blockchain_config(state, self.config.validate_external_messages) + { self.blockchain_config.store(config); } @@ -1124,7 +1243,10 @@ impl Drop for Inner { } } -fn load_blockchain_config(mc_state: &ShardStateStuff) -> Option> { +fn load_blockchain_config( + mc_state: &ShardStateStuff, + parse_executor_config: bool, +) -> Option> { let extra = mc_state.state_extra().ok()?; // Fill config. @@ -1141,10 +1263,26 @@ fn load_blockchain_config(mc_state: &ShardStateStuff) -> Option Some(Arc::new(config)), + Err(e) => { + tracing::error!( + block_id = %mc_state.block_id(), + "failed to parse config for executor: {e:?}", + ); + None + } + } + } else { + None + }; + Some(Arc::new(LatestBlockchainConfig { raw: extra.config.params.clone(), unpacked, modifiers, + parsed_config, })) } Err(e) => { @@ -1319,6 +1457,8 @@ pub enum RpcStateError { NotReady, #[error("not supported")] NotSupported, + #[error("message not accepted, exit code: {exit_code}")] + NotAccepted { exit_code: i32 }, #[error("internal: {0}")] Internal(#[from] anyhow::Error), #[error(transparent)] diff --git a/rpc/src/util/error_codes.rs b/rpc/src/util/error_codes.rs index 1e5409558b..07c54dd859 100644 --- a/rpc/src/util/error_codes.rs +++ b/rpc/src/util/error_codes.rs @@ -5,6 +5,7 @@ pub const INVALID_BOC_CODE: i32 = -32003; pub const TOO_LARGE_LIMIT_CODE: i32 = -32004; pub const TIMEOUT_CODE: i32 = -32005; pub const BAD_RESPONSE_CODE: i32 = -32006; +pub const INVALID_MESSAGE_CODE: i32 = -32007; pub const PARSE_ERROR_CODE: i32 = -32700; pub const INVALID_REQUEST_CODE: i32 = -32600;