feat: fetch and verify unsafe signer address (#424)

* feat: verify signer on client start * feat: add proof endpoint to consensus server * feat: run signer verify and advance async * feat: impl shutdown for eth consensus client * chore: fmt * fix: review changes * feat: add cli options for eth consensus client * nit: small doc change * fix: add eth_network to config * fix: add to chain_config * fix: use block hash to query proof * add Network to builder
a16z · Nov 24, 2024 · 0f9ea70 · 0f9ea70
1 parent fa22bfd
commit 0f9ea70
Show file tree

Hide file tree

Showing 17 changed files with 313 additions and 78 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/cli/src/main.rs b/cli/src/main.rs
@@ -7,24 +7,24 @@ use std::{
     sync::{Arc, Mutex},
 };
 
+use alloy::primitives::hex;
 use alloy::primitives::B256;
 use clap::{Args, Parser, Subcommand};
 use dirs::home_dir;
 use eyre::Result;
 use figment::providers::Serialized;
 use figment::value::Value;
 use futures::executor::block_on;
-use tracing::{error, info};
-use tracing_subscriber::filter::{EnvFilter, LevelFilter};
-use tracing_subscriber::FmtSubscriber;
-
 use helios_core::client::Client;
 use helios_core::consensus::Consensus;
 use helios_core::network_spec::NetworkSpec;
 use helios_ethereum::config::{cli::CliConfig, Config as EthereumConfig};
 use helios_ethereum::database::FileDB;
 use helios_ethereum::{EthereumClient, EthereumClientBuilder};
 use helios_opstack::{config::Config as OpStackConfig, OpStackClient, OpStackClientBuilder};
+use tracing::{error, info};
+use tracing_subscriber::filter::{EnvFilter, LevelFilter};
+use tracing_subscriber::FmtSubscriber;
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -191,6 +191,20 @@ struct OpStackArgs {
     execution_rpc: Option<String>,
     #[clap(short, long, env)]
     consensus_rpc: Option<String>,
+    #[clap(
+        short = 'w',
+        long = "ethereum-checkpoint",
+        env = "ETHEREUM_CHECKPOINT",
+        help = "Set custom weak subjectivity checkpoint for chosen Ethereum network. Helios uses this to sync and trustlessly fetch the correct unsafe signer address used by <NETWORK>"
+    )]
+    checkpoint: Option<B256>,
+    #[clap(
+        short = 'l',
+        long = "ethereum-load-external-fallback",
+        env = "ETHEREUM_LOAD_EXTERNAL_FALLBACK",
+        help = "Enable fallback for weak subjectivity checkpoint. Use if --ethereum-checkpoint fails."
+    )]
+    load_external_fallback: bool,
 }
 
 impl OpStackArgs {
@@ -232,6 +246,14 @@ impl OpStackArgs {
             user_dict.insert("rpc_port", Value::from(port));
         }
 
+        if self.load_external_fallback {
+            user_dict.insert("load_external_fallback", Value::from(true));
+        }
+
+        if let Some(checkpoint) = self.checkpoint {
+            user_dict.insert("checkpoint", Value::from(hex::encode(checkpoint)));
+        }
+
         Serialized::from(user_dict, &self.network)
     }
 }

diff --git a/core/src/execution/mod.rs b/core/src/execution/mod.rs
@@ -23,12 +23,11 @@ use self::types::Account;
 pub mod constants;
 pub mod errors;
 pub mod evm;
+pub mod proof;
 pub mod rpc;
 pub mod state;
 pub mod types;
 
-mod proof;
-
 #[derive(Clone)]
 pub struct ExecutionClient<N: NetworkSpec, R: ExecutionRpc<N>> {
     pub rpc: R,
@@ -64,7 +63,7 @@ impl<N: NetworkSpec, R: ExecutionRpc<N>> ExecutionClient<N, R> {
 
         let proof = self
             .rpc
-            .get_proof(address, slots, block.number.to())
+            .get_proof(address, slots, block.number.into())
             .await?;
 
         let account_path = keccak256(address).to_vec();

diff --git a/core/src/execution/rpc/http_rpc.rs b/core/src/execution/rpc/http_rpc.rs
@@ -54,12 +54,12 @@ impl<N: NetworkSpec> ExecutionRpc<N> for HttpRpc<N> {
         &self,
         address: Address,
         slots: &[B256],
-        block: u64,
+        block: BlockId,
     ) -> Result<EIP1186AccountProofResponse> {
         let proof_response = self
             .provider
             .get_proof(address, slots.to_vec())
-            .block_id(block.into())
+            .block_id(block)
             .await
             .map_err(|e| RpcError::new("get_proof", e))?;
 

diff --git a/core/src/execution/rpc/mock_rpc.rs b/core/src/execution/rpc/mock_rpc.rs
@@ -1,7 +1,9 @@
 use std::{fs::read_to_string, path::PathBuf};
 
 use alloy::primitives::{Address, B256, U256};
-use alloy::rpc::types::{AccessList, EIP1186AccountProofResponse, FeeHistory, Filter, Log};
+use alloy::rpc::types::{
+    AccessList, BlockId, EIP1186AccountProofResponse, FeeHistory, Filter, Log,
+};
 use async_trait::async_trait;
 use eyre::{eyre, Result};
 
@@ -26,7 +28,7 @@ impl<N: NetworkSpec> ExecutionRpc<N> for MockRpc {
         &self,
         _address: Address,
         _slots: &[B256],
-        _block: u64,
+        _block: BlockId,
     ) -> Result<EIP1186AccountProofResponse> {
         let proof = read_to_string(self.path.join("proof.json"))?;
         Ok(serde_json::from_str(&proof)?)

diff --git a/core/src/execution/rpc/mod.rs b/core/src/execution/rpc/mod.rs
@@ -1,5 +1,7 @@
 use alloy::primitives::{Address, B256, U256};
-use alloy::rpc::types::{AccessList, EIP1186AccountProofResponse, FeeHistory, Filter, Log};
+use alloy::rpc::types::{
+    AccessList, BlockId, EIP1186AccountProofResponse, FeeHistory, Filter, Log,
+};
 use async_trait::async_trait;
 use eyre::Result;
 
@@ -20,7 +22,7 @@ pub trait ExecutionRpc<N: NetworkSpec>: Send + Clone + Sync + 'static {
         &self,
         address: Address,
         slots: &[B256],
-        block: u64,
+        block: BlockId,
     ) -> Result<EIP1186AccountProofResponse>;
 
     async fn create_access_list(

diff --git a/core/src/lib.rs b/core/src/lib.rs
@@ -1,8 +1,7 @@
 pub mod client;
 pub mod consensus;
 pub mod errors;
+pub mod execution;
 pub mod network_spec;
 pub mod time;
 pub mod types;
-
-mod execution;
diff --git a/core/src/types.rs b/core/src/types.rs
@@ -25,6 +25,7 @@ pub struct Block<T: TransactionResponse> {
     pub size: U64,
     pub state_root: B256,
     pub timestamp: U64,
+    #[serde(default)]
     pub total_difficulty: U64,
     pub transactions: Transactions<T>,
     pub transactions_root: B256,

diff --git a/ethereum/src/config/mod.rs b/ethereum/src/config/mod.rs
@@ -91,3 +91,24 @@ impl Config {
         }
     }
 }
+
+impl From<BaseConfig> for Config {
+    fn from(base: BaseConfig) -> Self {
+        Config {
+            rpc_bind_ip: Some(base.rpc_bind_ip),
+            rpc_port: Some(base.rpc_port),
+            consensus_rpc: base.consensus_rpc.unwrap_or_default(),
+            execution_rpc: String::new(),
+            checkpoint: None,
+            default_checkpoint: base.default_checkpoint,
+            chain: base.chain,
+            forks: base.forks,
+            max_checkpoint_age: base.max_checkpoint_age,
+            data_dir: base.data_dir,
+            fallback: None,
+            load_external_fallback: base.load_external_fallback,
+            strict_checkpoint_age: base.strict_checkpoint_age,
+            database_type: None,
+        }
+    }
+}
diff --git a/ethereum/src/consensus.rs b/ethereum/src/consensus.rs
@@ -42,6 +42,7 @@ pub struct ConsensusClient<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> {
     pub block_recv: Option<Receiver<Block<Transaction>>>,
     pub finalized_block_recv: Option<watch::Receiver<Option<Block<Transaction>>>>,
     pub checkpoint_recv: watch::Receiver<Option<B256>>,
+    shutdown_send: watch::Sender<bool>,
     genesis_time: u64,
     config: Arc<Config>,
     phantom: PhantomData<(S, R, DB)>,
@@ -79,6 +80,7 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> Consensus<Transaction>
     }
 
     fn shutdown(&self) -> Result<()> {
+        self.shutdown_send.send(true)?;
         Ok(())
     }
 }
@@ -88,6 +90,7 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> ConsensusClient<S, R, D
         let (block_send, block_recv) = channel(256);
         let (finalized_block_send, finalized_block_recv) = watch::channel(None);
         let (checkpoint_send, checkpoint_recv) = watch::channel(None);
+        let (shutdown_send, shutdown_recv) = watch::channel(false);
 
         let config_clone = config.clone();
         let rpc = rpc.to_string();
@@ -103,6 +106,8 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> ConsensusClient<S, R, D
         #[cfg(target_arch = "wasm32")]
         let run = wasm_bindgen_futures::spawn_local;
 
+        let mut shutdown_rx = shutdown_recv.clone();
+
         run(async move {
             let mut inner = Inner::<S, R>::new(
                 &rpc,
@@ -138,28 +143,42 @@ impl<S: ConsensusSpec, R: ConsensusRpc<S>, DB: Database> ConsensusClient<S, R, D
             let mut interval = interval_at(start, std::time::Duration::from_secs(12));
 
             loop {
-                interval.tick().await;
-
-                let res = inner.advance().await;
-                if let Err(err) = res {
-                    warn!(target: "helios::consensus", "advance error: {}", err);
-                    continue;
-                }
-
-                let res = inner.send_blocks().await;
-                if let Err(err) = res {
-                    warn!(target: "helios::consensus", "send error: {}", err);
-                    continue;
+                tokio::select! {
+                    _ = shutdown_rx.changed() => {
+                        if *shutdown_rx.borrow() {
+                            info!(target: "helios::consensus", "shutting down consensus client");
+                            break;
+                        }
+                    }
+                    _ = interval.tick() => {
+                        let res = inner.advance().await;
+                        if let Err(err) = res {
+                            warn!(target: "helios::consensus", "advance error: {}", err);
+                            continue;
+                        }
+
+                        let res = inner.send_blocks().await;
+                        if let Err(err) = res {
+                            warn!(target: "helios::consensus", "send error: {}", err);
+                            continue;
+                        }
+                    }
                 }
             }
         });
 
-        save_new_checkpoints(checkpoint_recv.clone(), db.clone(), initial_checkpoint);
+        save_new_checkpoints(
+            checkpoint_recv.clone(),
+            db.clone(),
+            initial_checkpoint,
+            shutdown_recv,
+        );
 
         Ok(ConsensusClient {
             block_recv: Some(block_recv),
             finalized_block_recv: Some(finalized_block_recv),
             checkpoint_recv,
+            shutdown_send,
             genesis_time,
             config: config_clone,
             phantom: PhantomData,
@@ -177,6 +196,7 @@ fn save_new_checkpoints<DB: Database>(
     mut checkpoint_recv: watch::Receiver<Option<B256>>,
     db: Arc<DB>,
     initial_checkpoint: B256,
+    mut shutdown_recv: watch::Receiver<bool>,
 ) {
     #[cfg(not(target_arch = "wasm32"))]
     let run = tokio::spawn;
@@ -187,20 +207,28 @@ fn save_new_checkpoints<DB: Database>(
     run(async move {
         let mut last_saved_checkpoint = initial_checkpoint;
         loop {
-            let new_checkpoint = *checkpoint_recv.borrow_and_update();
-            if let Some(new_checkpoint) = new_checkpoint.as_ref() {
-                if *new_checkpoint != last_saved_checkpoint {
-                    // There is a more recent checkpoint to save
-                    if db.save_checkpoint(*new_checkpoint).is_err() {
-                        warn!(target: "helios::consensus", "failed to save checkpoint");
-                    } else {
-                        info!(target: "helios::consensus", "saved checkpoint to DB: 0x{}", hex::encode(*new_checkpoint));
-                        last_saved_checkpoint = *new_checkpoint;
+            tokio::select! {
+                _ = shutdown_recv.changed() => {
+                    if *shutdown_recv.borrow() {
+                        break;
+                    }
+                }
+                checkpoint_result = checkpoint_recv.changed() => {
+                    if checkpoint_result.is_err() {
+                        break;
+                    }
+                    let new_checkpoint = *checkpoint_recv.borrow_and_update();
+                    if let Some(new_checkpoint) = new_checkpoint.as_ref() {
+                        if *new_checkpoint != last_saved_checkpoint {
+                            if db.save_checkpoint(*new_checkpoint).is_err() {
+                                warn!(target: "helios::consensus", "failed to save checkpoint");
+                            } else {
+                                info!(target: "helios::consensus", "saved checkpoint to DB: 0x{}", hex::encode(*new_checkpoint));
+                                last_saved_checkpoint = *new_checkpoint;
+                            }
+                        }
                     }
                 }
-            }
-            if checkpoint_recv.changed().await.is_err() {
-                break;
             }
         }
     });

diff --git a/helios-ts/src/opstack.rs b/helios-ts/src/opstack.rs
@@ -43,6 +43,8 @@ impl OpStackClient {
             consensus_rpc,
             chain: network_config.chain,
             rpc_socket: None,
+            load_external_fallback: None,
+            checkpoint: None,
         };
 
         let inner = map_err(OpStackClientBuilder::new().config(config).build())?;

diff --git a/opstack/Cargo.toml b/opstack/Cargo.toml
@@ -37,6 +37,8 @@ snap = "1"
 figment = { version = "0.10.7", features = ["toml", "env"] }
 
 helios-core = { path = "../core" }
+helios-ethereum = { path = "../ethereum" }
+helios-consensus-core = { path = "../ethereum/consensus-core" }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 # server

diff --git a/opstack/bin/server.rs b/opstack/bin/server.rs
@@ -21,16 +21,20 @@ async fn main() -> Result<()> {
 
     let chain_id = config.chain.chain_id;
     let unsafe_signer = config.chain.unsafe_signer;
+    let system_config_contract = config.chain.system_config_contract;
     let server_addr = cli.server_address;
     let gossip_addr = cli.gossip_address;
     let replica_urls = cli.replica_urls.unwrap_or_default();
+    let execution_rpc = cli.execution_rpc;
 
     start_server(
         server_addr,
         gossip_addr,
         chain_id,
         unsafe_signer,
+        system_config_contract,
         replica_urls,
+        execution_rpc,
     )
     .await?;
 
@@ -67,4 +71,6 @@ struct Cli {
     gossip_address: SocketAddr,
     #[clap(short, long, value_delimiter = ',')]
     replica_urls: Option<Vec<Url>>,
+    #[clap(short, long)]
+    execution_rpc: Url,
 }