fix(l2): validate signature p2p based

cmd/ethrex/l2/command.rs

@@ -23,7 +23,9 @@ use ethrex_l2_sdk::call_contract;
 use ethrex_p2p::network::peer_table;
 use ethrex_p2p::rlpx::l2::l2_connection::P2PBasedContext;
 use ethrex_rpc::{
-    EthClient, clients::beacon::BeaconClient, types::block_identifier::BlockIdentifier,
+    EthClient,
+    clients::{beacon::BeaconClient, eth::eth_sender::LeadSequencerVerifierStruct},
+    types::block_identifier::BlockIdentifier,
 };
 use ethrex_storage::{EngineType, Store, UpdateBatch};
 use ethrex_storage_rollup::{EngineTypeRollup, StoreRollup};
@@ -183,6 +185,9 @@ impl Command {
                         error!("{err}");
                     })?;
 
+                let eth_client =
+                    EthClient::new_with_multiple_urls(l2_sequencer_cfg.eth.rpc_url.clone())?;
+
                 // TODO: This should be handled differently, the current problem
                 // with using opts.node_opts.p2p_enabled is that with the removal
                 // of the l2 feature flag, p2p_enabled is set to true by default
@@ -202,6 +207,12 @@ impl Command {
                         Some(P2PBasedContext {
                             store_rollup: rollup_store.clone(),
                             committer_key: Arc::new(l2_sequencer_cfg.l1_committer.l1_private_key),
+                            lead_sequencer_verifier: Arc::new(LeadSequencerVerifierStruct {
+                                on_chain_proposer: l2_sequencer_cfg
+                                    .l1_committer
+                                    .on_chain_proposer_address,
+                                eth_client,
+                            }),
                         }),
                     )
                     .await;

cmd/ethrex/l2/initializers.rs

@@ -1,15 +1,18 @@
+use std::collections::HashMap;
 use std::fs::read_to_string;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 
 use ethrex_blockchain::Blockchain;
 use ethrex_common::Address;
+use ethrex_l2_rpc::RpcApiContext as L2RpcApiContext;
 use ethrex_p2p::kademlia::KademliaTable;
 use ethrex_p2p::peer_handler::PeerHandler;
 use ethrex_p2p::sync_manager::SyncManager;
 use ethrex_p2p::types::{Node, NodeRecord};
+use ethrex_rpc::{GasTipEstimator, NodeData};
 use ethrex_storage::Store;
 use ethrex_storage_rollup::{EngineTypeRollup, StoreRollup};
-use tokio::sync::Mutex;
+use tokio::sync::Mutex as TokioMutex;
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use tracing::warn;
@@ -26,7 +29,7 @@ use crate::utils::{get_client_version, read_jwtsecret_file};
 pub async fn init_rpc_api(
     opts: &L1Options,
     l2_opts: &L2Options,
-    peer_table: Arc<Mutex<KademliaTable>>,
+    peer_table: Arc<TokioMutex<KademliaTable>>,
     local_p2p_node: Node,
     local_node_record: NodeRecord,
     store: Store,
@@ -47,20 +50,34 @@ pub async fn init_rpc_api(
     )
     .await;
 
+    // TODO: Refactor how filters are handled,
+    // filters are used by the filters endpoints (eth_newFilter, eth_getFilterChanges, ...etc)
+    let active_filters = Arc::new(Mutex::new(HashMap::new()));
+    let service_context = L2RpcApiContext {
+        l1_ctx: ethrex_rpc::RpcApiContext {
+            storage: store.clone(),
+            blockchain: blockchain.clone(),
+            active_filters: active_filters.clone(),
+            syncer: Arc::new(syncer),
+            peer_handler,
+            node_data: NodeData {
+                jwt_secret: read_jwtsecret_file(&opts.authrpc_jwtsecret),
+                local_p2p_node,
+                local_node_record,
+                client_version: get_client_version(),
+            },
+            gas_tip_estimator: Arc::new(TokioMutex::new(GasTipEstimator::new())),
+        },
+        valid_delegation_addresses: get_valid_delegation_addresses(l2_opts),
+        sponsor_pk: l2_opts.sponsor_private_key,
+        rollup_store,
+    };
+
     let rpc_api = ethrex_l2_rpc::start_api(
         get_http_socket_addr(opts),
         get_authrpc_socket_addr(opts),
-        store,
-        blockchain,
-        read_jwtsecret_file(&opts.authrpc_jwtsecret),
-        local_p2p_node,
-        local_node_record,
-        syncer,
-        peer_handler,
-        get_client_version(),
-        get_valid_delegation_addresses(l2_opts),
-        l2_opts.sponsor_private_key,
-        rollup_store,
+        service_context,
+        active_filters,
     );
 
     tracker.spawn(rpc_api);

crates/l2/based/block_fetcher.rs

@@ -88,6 +88,7 @@ impl BlockFetcherState {
         sequencer_state: SequencerState,
     ) -> Result<Self, BlockFetcherError> {
         let eth_client = EthClient::new_with_multiple_urls(cfg.eth.rpc_url.clone())?;
+        dbg!(cfg.eth.rpc_url.clone());
         let last_l1_block_fetched = eth_client
             .get_last_fetched_l1_block(cfg.l1_watcher.bridge_address)
             .await?

crates/l2/networking/rpc/Cargo.toml

@@ -9,7 +9,6 @@ edition = "2024"
 ethrex-common.workspace = true
 ethrex-storage.workspace = true
 ethrex-blockchain.workspace = true
-ethrex-p2p.workspace = true
 ethrex-storage-rollup.workspace = true
 ethrex-l2-common.workspace = true
 ethrex-rpc.workspace = true

crates/l2/networking/rpc/lib.rs

@@ -4,4 +4,5 @@ mod rpc;
 pub mod signer;
 pub mod utils;
 
+pub use rpc::RpcApiContext;
 pub use rpc::start_api;

crates/l2/networking/rpc/rpc.rs

@@ -3,29 +3,15 @@ use crate::l2::l1_message::GetL1MessageProof;
 use crate::utils::{RpcErr, RpcNamespace, resolve_namespace};
 use axum::extract::State;
 use axum::{Json, Router, http::StatusCode, routing::post};
-use bytes::Bytes;
-use ethrex_blockchain::Blockchain;
 use ethrex_common::types::Transaction;
-use ethrex_p2p::peer_handler::PeerHandler;
-use ethrex_p2p::sync_manager::SyncManager;
-use ethrex_p2p::types::Node;
-use ethrex_p2p::types::NodeRecord;
-use ethrex_rpc::RpcHandler as L1RpcHandler;
+use ethrex_rpc::{ActiveFilters, RpcHandler as L1RpcHandler, RpcRequestWrapper};
 use ethrex_rpc::{
-    GasTipEstimator, NodeData, RpcRequestWrapper,
     types::transaction::SendRawTransactionRequest,
     utils::{RpcRequest, RpcRequestId},
 };
-use ethrex_storage::Store;
 use serde_json::Value;
-use std::{
-    collections::HashMap,
-    future::IntoFuture,
-    net::SocketAddr,
-    sync::{Arc, Mutex},
-    time::Duration,
-};
-use tokio::{net::TcpListener, sync::Mutex as TokioMutex};
+use std::{future::IntoFuture, net::SocketAddr, time::Duration};
+use tokio::net::TcpListener;
 use tower_http::cors::CorsLayer;
 use tracing::{debug, info};
 
@@ -61,44 +47,14 @@ pub const FILTER_DURATION: Duration = {
     }
 };
 
-#[expect(clippy::too_many_arguments)]
 pub async fn start_api(
     http_addr: SocketAddr,
     authrpc_addr: SocketAddr,
-    storage: Store,
-    blockchain: Arc<Blockchain>,
-    jwt_secret: Bytes,
-    local_p2p_node: Node,
-    local_node_record: NodeRecord,
-    syncer: SyncManager,
-    peer_handler: PeerHandler,
-    client_version: String,
-    valid_delegation_addresses: Vec<Address>,
-    sponsor_pk: SecretKey,
-    rollup_store: StoreRollup,
+    service_context: RpcApiContext,
+    active_filters: ActiveFilters,
 ) -> Result<(), RpcErr> {
     // TODO: Refactor how filters are handled,
     // filters are used by the filters endpoints (eth_newFilter, eth_getFilterChanges, ...etc)
-    let active_filters = Arc::new(Mutex::new(HashMap::new()));
-    let service_context = RpcApiContext {
-        l1_ctx: ethrex_rpc::RpcApiContext {
-            storage,
-            blockchain,
-            active_filters: active_filters.clone(),
-            syncer: Arc::new(syncer),
-            peer_handler,
-            node_data: NodeData {
-                jwt_secret,
-                local_p2p_node,
-                local_node_record,
-                client_version,
-            },
-            gas_tip_estimator: Arc::new(TokioMutex::new(GasTipEstimator::new())),
-        },
-        valid_delegation_addresses,
-        sponsor_pk,
-        rollup_store,
-    };
 
     // Periodically clean up the active filters for the filters endpoints.
     tokio::task::spawn(async move {

crates/networking/p2p-common/Cargo.toml

@@ -0,0 +1,4 @@
+[package]
+name = "ethrex-p2p-common"
+version = "0.1.0"
+edition = "2024"

crates/networking/p2p/Cargo.toml

@@ -26,6 +26,7 @@ secp256k1.workspace = true
 spawned-rt.workspace = true
 spawned-concurrency.workspace = true
 keccak-hash.workspace = true
+async-trait.workspace = true
 
 tokio-stream = "0.1.17"
 futures = "0.3.31"

crates/networking/p2p/rlpx/l2/l2_connection.rs

@@ -26,12 +26,18 @@ pub struct L2ConnectedState {
     pub committer_key: Arc<SecretKey>,
     pub next_block_broadcast: Instant,
     pub next_batch_broadcast: Instant,
+    pub lead_sequencer_verifier: Arc<dyn LeadSequencerVerifier>,
 }
 
+#[async_trait::async_trait]
+pub trait LeadSequencerVerifier: std::fmt::Debug + Send + Sync {
+    fn validate_signature(&self, recovered_lead_sequencer: Address) -> bool;
+}
 #[derive(Debug, Clone)]
 pub struct P2PBasedContext {
     pub store_rollup: StoreRollup,
     pub committer_key: Arc<SecretKey>,
+    pub lead_sequencer_verifier: Arc<dyn LeadSequencerVerifier>,
 }
 
 #[derive(Debug, Clone)]
@@ -76,6 +82,7 @@ impl L2ConnState {
                     committer_key: ctxt.committer_key.clone(),
                     next_block_broadcast: Instant::now() + PERIODIC_BLOCK_BROADCAST_INTERVAL,
                     next_batch_broadcast: Instant::now() + PERIODIC_BATCH_BROADCAST_INTERVAL,
+                    lead_sequencer_verifier: ctxt.lead_sequencer_verifier.clone(),
                 };
                 *self = L2ConnState::Connected(state);
                 Ok(())
@@ -85,11 +92,6 @@ impl L2ConnState {
     }
 }
 
-fn validate_signature(_recovered_lead_sequencer: Address) -> bool {
-    // Until the RPC module can be included in the P2P crate, we skip the validation
-    true
-}
-
 pub(crate) async fn handle_based_capability_message(
     established: &mut Established,
     msg: L2Message,
@@ -215,7 +217,10 @@ async fn should_process_new_block(
         RLPxError::CryptographyError(e.to_string())
     })?;
 
-    if !validate_signature(recovered_lead_sequencer) {
+    if !l2_state
+        .lead_sequencer_verifier
+        .validate_signature(recovered_lead_sequencer)
+    {
         return Ok(false);
     }
     let mut signature = [0u8; 68];
@@ -268,7 +273,10 @@ async fn should_process_batch_sealed(
             RLPxError::CryptographyError(e.to_string())
         })?;
 
-    if !validate_signature(recovered_lead_sequencer) {
+    if !l2_state
+        .lead_sequencer_verifier
+        .validate_signature(recovered_lead_sequencer)
+    {
         return Ok(false);
     }
 

crates/networking/rpc/clients/eth/eth_sender.rs

@@ -10,7 +10,12 @@ use bytes::Bytes;
 use ethrex_common::H256;
 use ethrex_common::types::{GenericTransaction, TxKind};
 use ethrex_common::{Address, U256};
+use ethrex_p2p::rlpx::l2::l2_connection::LeadSequencerVerifier;
+use ethrex_rlp::encode::RLPEncode;
+use keccak_hash::keccak;
+use secp256k1::SecretKey;
 use serde_json::json;
+use tokio::task::block_in_place;
 
 #[derive(Default, Clone, Debug)]
 pub struct Overrides {
@@ -27,6 +32,30 @@ pub struct Overrides {
     pub block: Option<BlockIdentifier>,
 }
 
+#[derive(Debug, Clone)]
+pub struct LeadSequencerVerifierStruct {
+    pub on_chain_proposer: Address,
+    pub eth_client: EthClient,
+}
+impl LeadSequencerVerifier for LeadSequencerVerifierStruct {
+    fn validate_signature(&self, recovered_lead_sequencer: Address) -> bool {
+        // Implement the signature validation logic here
+        block_in_place(|| {
+            tokio::runtime::Runtime::new()
+                .expect("Failed to create runtime")
+                .block_on(async {
+                    let current_lead_sequencer: Address = self
+                        .eth_client
+                        .call(self.on_chain_proposer, Bytes::new(), Overrides::default())
+                        .await
+                        .expect("Failed to call on_chain_proposer")
+                        .parse()
+                        .expect("Failed to parse current lead sequencer address");
+                    current_lead_sequencer == recovered_lead_sequencer
+                })
+        })
+    }
+}
 impl EthClient {
     pub async fn call(
         &self,