fix(security): Security audit, fix security findings in grevm parallel EVM (#99)

follow #97

---------

Co-authored-by: Richard <xunqiu2@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: nekomoto911 <nekomoto911@gmail.com>

Author: 4 people

Cargo.toml

@@ -29,6 +29,9 @@ auto_impl = "1.3"
 lazy_static = "1.5.0"
 dashmap = "6.0"
 
+# logging
+tracing = "0.1"
+
 # metrics
 metrics = "0.24"
 metrics-derive = "0.1"

benches/gigagas.rs

@@ -424,7 +424,9 @@ fn bench_dependent_erc20(c: &mut Criterion, db_latency_us: u64, num_eoa: usize,
 
 fn bench_uniswap(c: &mut Criterion, db_latency_us: u64) {
     let block_size = (GIGA_GAS as f64 / uniswap::ESTIMATED_GAS_USED as f64).ceil() as usize;
-    let mut final_state = HashMap::from([account::mock_miner_account()]);
+    let mut final_state = HashMap::default();
+    let miner = account::mock_miner_account();
+    final_state.insert(miner.0, miner.1);
     let mut final_bytecodes = HashMap::default();
     let mut final_txs = Vec::<TxEnv>::new();
     for _ in 0..block_size {

src/async_commit.rs

@@ -6,7 +6,25 @@ use revm_context::{
 use revm_primitives::Address;
 
 use crate::{GrevmError, ParallelState, TxId};
-use std::cmp::Ordering;
+use std::{cell::UnsafeCell, cmp::Ordering};
+
+/// Only constructible within the commit thread's scope (or sequential fallback).
+/// Guarantees exclusive mutable access to ParallelState.
+pub(crate) struct CommitGuard<'a, DB: DatabaseRef> {
+    state: &'a UnsafeCell<ParallelState<DB>>,
+}
+
+impl<'a, DB: DatabaseRef> CommitGuard<'a, DB> {
+    pub(crate) fn new(state: &'a UnsafeCell<ParallelState<DB>>) -> Self {
+        Self { state }
+    }
+
+    pub(crate) fn state_mut(&mut self) -> &mut ParallelState<DB> {
+        // SAFETY: CommitGuard requires `&mut self`, ensuring exclusive access
+        // to the underlying state for the guard's owner.
+        unsafe { &mut *self.state.get() }
+    }
+}
 
 /// `StateAsyncCommit` asynchronously finalizes transaction states,
 /// serving two critical purposes:
@@ -20,35 +38,44 @@ where
 {
     coinbase: Address,
     results: Vec<ExecutionResult>,
-    state: &'a ParallelState<DB>,
+    guard: CommitGuard<'a, DB>,
     commit_result: Result<(), GrevmError<DB::Error>>,
     disable_nonce_check: bool,
 }
 
+// SAFETY: StateAsyncCommit is only used within a single commit thread (never shared).
+// CommitGuard ensures exclusive mutable access through its `&mut self` requirement.
+unsafe impl<DB: DatabaseRef + Send + Sync> Send for StateAsyncCommit<'_, DB> where DB::Error: Send {}
+
 impl<'a, DB> StateAsyncCommit<'a, DB>
 where
     DB: DatabaseRef,
 {
     pub(crate) fn new(
         coinbase: Address,
-        state: &'a ParallelState<DB>,
+        guard: CommitGuard<'a, DB>,
         disable_nonce_check: bool,
     ) -> Self {
-        Self { coinbase, results: vec![], state, commit_result: Ok(()), disable_nonce_check }
+        Self { coinbase, results: vec![], guard, commit_result: Ok(()), disable_nonce_check }
     }
 
-    fn state_mut(&self) -> &mut ParallelState<DB> {
-        #[allow(invalid_reference_casting)]
-        unsafe {
-            &mut *(self.state as *const ParallelState<DB> as *mut ParallelState<DB>)
-        }
+    pub(crate) fn state_mut(&mut self) -> &mut ParallelState<DB> {
+        self.guard.state_mut()
+    }
+
+    /// SAFETY: Shared read access to state is safe because `ParallelState` reads go through
+    /// `DashMap` (thread-safe) or the immutable database.
+    fn state_ref(&self) -> &ParallelState<DB> {
+        // Safe to get a shared reference since we only mutate exclusively when we have `&mut self`
+        unsafe { &*self.guard.state.get() }
     }
 
     pub(crate) fn init(&mut self) -> Result<(), DB::Error> {
         // Accesses the coinbase account to ensure proper handling of miner rewards (via
         // increment_balances) within ParallelState. This preemptive access guarantees correct state
         // synchronization when applying miner rewards during the final commitment phase.
-        match self.state_mut().basic(self.coinbase) {
+        let coinbase = self.coinbase;
+        match self.state_mut().basic(coinbase) {
             Ok(_) => Ok(()),
             Err(e) => Err(e),
         }
@@ -70,7 +97,7 @@ where
         // integrity and prevent double-spending attacks.
         let ResultAndState { result, state, lazy_reward } = result_and_state;
         if !self.disable_nonce_check {
-            match self.state.basic_ref(tx_env.caller) {
+            match self.state_ref().basic_ref(tx_env.caller) {
                 Ok(info) => {
                     if let Some(info) = info {
                         let expect = info.nonce;
@@ -108,6 +135,9 @@ where
             }
         }
         self.results.push(result);
+        if self.commit_result.is_err() {
+            return;
+        }
         self.state_mut().commit(state);
 
         // In Ethereum, each transaction includes a miner reward, which would introduce write
@@ -117,6 +147,7 @@ where
         // correct concurrency - even if subsequent transactions access the miner's account, they
         // will read the proper miner state from ParallelState (verified via commit_idx) without
         // creating artificial dependencies.
-        assert!(self.state_mut().increment_balances(vec![(self.coinbase, lazy_reward)]).is_ok());
+        let coinbase = self.coinbase;
+        assert!(self.state_mut().increment_balances(vec![(coinbase, lazy_reward)]).is_ok());
     }
 }

src/hint.rs

@@ -4,7 +4,7 @@ use revm::primitives::{
     Address, B256, Bytes, TxKind, U256, alloy_primitives::U160, keccak256, ruint::UintTryFrom,
 };
 use revm_context::TxEnv;
-use std::{cmp::max, sync::Arc};
+use std::{cell::UnsafeCell, cmp::max, sync::Arc};
 
 /// This module provides functionality for parsing and handling execution hints
 /// for parallel transaction execution in the context of Ethereum-like blockchains.
@@ -87,21 +87,25 @@ impl RWSet {
 /// Struct to hold shared transaction states and provide methods for parsing
 /// and handling execution hints for parallel transaction execution.
 pub(crate) struct ParallelExecutionHints {
-    rw_set: Vec<RWSet>,
+    rw_set: UnsafeCell<Vec<RWSet>>,
     txs: Arc<Vec<TxEnv>>,
 }
 
+// SAFETY: Each thread in fork_join_util writes to disjoint indices of rw_set.
+unsafe impl Sync for ParallelExecutionHints {}
+
 impl ParallelExecutionHints {
     pub(crate) fn new(txs: Arc<Vec<TxEnv>>) -> Self {
-        Self { rw_set: vec![RWSet::new(); txs.len()], txs }
+        Self { rw_set: UnsafeCell::new(vec![RWSet::new(); txs.len()]), txs }
     }
 
     fn generate_dependency(&self) -> TxDependency {
         let num_txs = self.txs.len();
         let mut last_write_tx: HashMap<LocationAndType, TxId> = HashMap::new();
         let mut dependent_tx: Vec<Option<TxId>> = vec![None; num_txs];
         let mut affect_txs = vec![HashSet::new(); num_txs];
-        for (txid, rw_set) in self.rw_set.iter().enumerate() {
+        let rw_set = unsafe { &*self.rw_set.get() };
+        for (txid, rw_set) in rw_set.iter().enumerate() {
             for location in rw_set.read_set.iter() {
                 if let Some(&previous) = last_write_tx.get(location) {
                     let new_dep = dependent_tx[txid].map_or(previous, |dep| max(dep, previous));
@@ -120,8 +124,7 @@ impl ParallelExecutionHints {
         let txs = self.txs.clone();
         // Utilize fork-join utility to process transactions in parallel
         fork_join_util(txs.len(), None, |start_tx, end_tx, _| {
-            #[allow(invalid_reference_casting)]
-            let hints = unsafe { &mut *(&self.rw_set as *const Vec<RWSet> as *mut Vec<RWSet>) };
+            let hints = unsafe { &mut *self.rw_set.get() };
             for index in start_tx..end_tx {
                 let tx_env = &txs[index];
                 let rw_set = &mut hints[index];
@@ -281,16 +284,25 @@ impl ParallelExecutionHints {
         true
     }
 
-    fn get_contract_type(_contract_address: Address, _data: &Bytes) -> ContractType {
-        // TODO(gaoxin): Parse the correct contract type to determined how to handle call data
-        ContractType::ERC20
+    fn get_contract_type(_contract_address: Address, data: &Bytes) -> ContractType {
+        // Without bytecode analysis, we use a heuristic: if the function selector matches
+        // a known ERC20 function, treat it as ERC20. Otherwise, return UNKNOWN to be
+        // conservative and avoid incorrect dependency hints.
+        if data.len() >= 4 {
+            let func_id = u32::from_be_bytes([data[0], data[1], data[2], data[3]]);
+            match ERC20Function::from(func_id) {
+                ERC20Function::UNKNOWN => ContractType::UNKNOWN,
+                _ => ContractType::ERC20,
+            }
+        } else {
+            ContractType::UNKNOWN
+        }
     }
 
     fn decode_contract_parameters(data: &Bytes) -> (u32, Vec<B256>) {
-        let func_id: u32 = 0;
         let mut parameters: Vec<B256> = Vec::new();
         if data.len() <= 4 {
-            return (func_id, parameters);
+            return (0, parameters);
         }
 
         let func_id = u32::from_be_bytes([data[0], data[1], data[2], data[3]]);

src/lib.rs

@@ -179,6 +179,9 @@ pub fn fork_join_util<'scope, F>(num_elements: usize, num_partitions: Option<usi
 where
     F: Fn(usize, usize, usize) + Send + Sync + 'scope,
 {
+    if num_elements == 0 {
+        return;
+    }
     let parallel_cnt = num_partitions.unwrap_or(*CONCURRENT_LEVEL);
     let remaining = num_elements % parallel_cnt;
     let chunk_size = num_elements / parallel_cnt;

src/scheduler.rs

@@ -1,8 +1,12 @@
 use crate::{
     AbortReason, CONCURRENT_LEVEL, FALLBACK_SEQUENTIAL, GrevmError, LocationAndType, MemoryEntry,
     ParallelState, ReadVersion, Task, TransactionResult, TransactionStatus, TxId, TxState,
-    TxVersion, async_commit::StateAsyncCommit, hint::ParallelExecutionHints, storage::CacheDB,
-    tx_dependency::TxDependency, utils::ContinuousDetectSet,
+    TxVersion,
+    async_commit::{CommitGuard, StateAsyncCommit},
+    hint::ParallelExecutionHints,
+    storage::CacheDB,
+    tx_dependency::TxDependency,
+    utils::ContinuousDetectSet,
 };
 use ahash::{AHashMap as HashMap, AHashSet as HashSet};
 use alloy_evm::{
@@ -25,6 +29,7 @@ use revm_inspector::NoOpInspector;
 use revm_primitives::Address;
 
 use std::{
+    cell::UnsafeCell,
     cmp::max,
     collections::BTreeMap,
     fmt::Debug,
@@ -36,6 +41,9 @@ use std::{
     time::Instant,
 };
 
+/// min number of txs to parallel execute.
+const MIN_PARALLEL_TXS: usize = 64;
+
 pub(crate) type MVMemory = DashMap<LocationAndType, BTreeMap<TxId, MemoryEntry>>;
 
 #[derive(Metrics)]
@@ -261,7 +269,7 @@ where
     env: BlockEnv,
     block_size: usize,
     txs: Arc<Vec<TxEnv>>,
-    state: ParallelState<DB>,
+    state: UnsafeCell<ParallelState<DB>>,
     results: Mutex<Vec<ExecutionResult>>,
     tx_states: Vec<Mutex<TxState>>,
     tx_results: Vec<Mutex<Option<TransactionResult<DB::Error>>>>,
@@ -276,6 +284,12 @@ where
     metrics: ExecuteMetricsCollector,
 }
 
+// SAFETY: Scheduler is shared across threads via `thread::scope`. The `UnsafeCell<ParallelState>`
+// is safe because: (1) only the commit thread mutates it (via StateAsyncCommit), serialized by
+// finality ordering, (2) worker threads only read via DatabaseRef (DashMap, thread-safe),
+// (3) fallback_sequential() is only called after all threads have joined.
+unsafe impl<DB: DatabaseRef + Send + Sync> Sync for Scheduler<DB> where DB::Error: Send + Sync {}
+
 impl<DB> Debug for Scheduler<DB>
 where
     DB: DatabaseRef,
@@ -315,7 +329,7 @@ where
             env,
             block_size: num_txs,
             txs,
-            state,
+            state: UnsafeCell::new(state),
             results: Mutex::new(vec![]),
             tx_states: (0..num_txs).map(|_| Mutex::new(TxState::default())).collect(),
             tx_results: (0..num_txs).map(|_| Mutex::new(None)).collect(),
@@ -375,17 +389,14 @@ where
 
             if (Instant::now() - start).as_millis() > 8_000 {
                 start = Instant::now();
-                println!(
-                    "stuck..., block_number: {}, finality_idx: {}, validation_idx: {}, execution_idx: {}",
-                    self.env.number,
-                    self.scheduler_ctx.finality_idx(),
-                    self.scheduler_ctx.validation_idx(),
-                    self.scheduler_ctx.executed_set.continuous_idx(),
+                tracing::warn!(
+                    target: "grevm::scheduler",
+                    block_number = %self.env.number,
+                    finality_idx = self.scheduler_ctx.finality_idx(),
+                    validation_idx = self.scheduler_ctx.validation_idx(),
+                    execution_idx = self.scheduler_ctx.executed_set.continuous_idx(),
+                    "parallel execution stuck",
                 );
-                let status: Vec<(TxId, TransactionStatus)> =
-                    self.tx_states.iter().map(|s| s.lock().status.clone()).enumerate().collect();
-                println!("transaction status: {:?}", status);
-                self.tx_dependency.print();
             }
         }
     }
@@ -394,7 +405,7 @@ where
         let mut commit_idx = 0;
         let mut commiter = commiter.lock();
         let async_commit_state =
-            std::env::var("ASYNC_COMMIT_STATE").map_or(true, |s| s.parse().unwrap());
+            std::env::var("ASYNC_COMMIT_STATE").map_or(true, |s| s.parse().unwrap_or(true));
         while !self.abort.load(Ordering::Acquire) && commit_idx < self.block_size {
             while commit_idx < self.scheduler_ctx.finality_idx.load(Ordering::Acquire) {
                 if async_commit_state {
@@ -420,7 +431,7 @@ where
 
     /// Take `ExecutionResult` and `ParallelState`
     pub fn take_result_and_state(self) -> (Vec<ExecutionResult>, ParallelState<DB>) {
-        (self.results.into_inner(), self.state)
+        (self.results.into_inner(), self.state.into_inner())
     }
 
     /// Paralle execution
@@ -432,16 +443,18 @@ where
         self.metrics.total_tx_cnt.store(self.block_size, Ordering::Relaxed);
         let concurrency_level = concurrency_level.unwrap_or(
             std::env::var("GREVM_CONCURRENT_LEVEL")
-                .map_or(*CONCURRENT_LEVEL, |s| s.parse().unwrap()),
+                .map_or(*CONCURRENT_LEVEL, |s| s.parse().unwrap_or(*CONCURRENT_LEVEL)),
         );
-        if *FALLBACK_SEQUENTIAL {
+        if *FALLBACK_SEQUENTIAL || self.block_size < MIN_PARALLEL_TXS {
             return self.fallback_sequential();
         }
         let commiter = Mutex::new(StateAsyncCommit::new(
             self.env.beneficiary,
-            &self.state,
+            CommitGuard::new(&self.state),
             self.cfg.disable_nonce_check,
         ));
+
+        let state_ref = unsafe { &*self.state.get() };
         commiter.lock().init().map_err(|e| GrevmError { txid: 0, error: EVMError::Database(e) })?;
         thread::scope(|scope| {
             scope.spawn(|| {
@@ -458,7 +471,7 @@ where
                     let cache_db = CacheDB::new(
                         self.cfg.spec,
                         self.env.beneficiary,
-                        &self.state,
+                        state_ref,
                         &self.mv_memory,
                         &self.scheduler_ctx.commit_idx,
                     );
@@ -548,13 +561,6 @@ where
         Ok(())
     }
 
-    fn state_mut(&self) -> &mut ParallelState<DB> {
-        #[allow(invalid_reference_casting)]
-        unsafe {
-            &mut *(&self.state as *const ParallelState<DB> as *mut ParallelState<DB>)
-        }
-    }
-
     /// Fallback to sequential execution
     pub fn fallback_sequential(&self) -> Result<(), GrevmError<DB::Error>> {
         let mut results = self.results.lock();
@@ -564,7 +570,8 @@ where
         }
 
         let mut sequential_results = Vec::with_capacity(self.block_size - num_commit);
-        let state_mut = self.state_mut();
+        let mut commit_guard = CommitGuard::new(&self.state);
+        let state_mut = commit_guard.state_mut();
         {
             let evm = Context::mainnet()
                 .with_db(state_mut)