feat (batcher): allow whitelist of non paying specific address (#523)

Co-authored-by: Mariano Nicolini <mariano.nicolini.91@gmail.com>
Co-authored-by: Urix <43704209+uri-99@users.noreply.github.com>
Co-authored-by: JuArce <52429267+JuArce@users.noreply.github.com>
Co-authored-by: MauroFab <maurotoscano2@gmail.com>
Co-authored-by: Mauro Toscano <12560266+MauroToscano@users.noreply.github.com>

Author: 5 people

batcher/aligned-batcher/src/config/mod.rs

@@ -1,3 +1,4 @@
+use ethers::types::Address;
 use serde::Deserialize;
 
 #[derive(Clone, Debug, Deserialize)]
@@ -6,6 +7,12 @@ pub struct ECDSAConfig {
     pub private_key_store_password: String,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct NonPayingConfig {
+    pub address: Address,
+    pub replacement: Address,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct BatcherConfigFromYaml {
     pub block_interval: u64,
@@ -14,6 +21,7 @@ pub struct BatcherConfigFromYaml {
     pub max_batch_size: usize,
     pub eth_ws_reconnects: usize,
     pub pre_verification_is_enabled: bool,
+    pub non_paying: Option<NonPayingConfig>,
 }
 
 #[derive(Debug, Deserialize)]

batcher/aligned-batcher/src/lib.rs

@@ -5,32 +5,33 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::Duration;
 
-use crate::eth::BatchVerifiedEventStream;
-use aligned_batcher_lib::types::{
-    BatchInclusionData, ClientMessage, VerificationCommitmentBatch, VerificationData,
-    VerificationDataCommitment,
-};
 use aws_sdk_s3::client::Client as S3Client;
-use eth::{BatchVerifiedFilter, BatcherPaymentService};
 use ethers::prelude::{Middleware, Provider};
 use ethers::providers::Ws;
 use ethers::types::{Address, U256};
 use futures_util::stream::{self, SplitSink};
 use futures_util::{future, SinkExt, StreamExt, TryStreamExt};
 use lambdaworks_crypto::merkle_tree::merkle::MerkleTree;
-use log::{debug, error, info};
+use log::{debug, error, info, warn};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{Mutex, RwLock};
 use tokio::time::timeout;
 use tokio_tungstenite::tungstenite::error::ProtocolError;
 use tokio_tungstenite::tungstenite::protocol::{frame::coding::CloseCode, CloseFrame};
 use tokio_tungstenite::tungstenite::{Error, Message};
 use tokio_tungstenite::WebSocketStream;
+
+use aligned_batcher_lib::types::{
+    BatchInclusionData, ClientMessage, VerificationCommitmentBatch, VerificationData,
+    VerificationDataCommitment,
+};
+use eth::{BatchVerifiedFilter, BatcherPaymentService};
 use types::batch_queue::BatchQueue;
 use types::errors::BatcherError;
 
-use crate::config::{ConfigFromYaml, ContractDeploymentOutput};
+use crate::config::{ConfigFromYaml, ContractDeploymentOutput, NonPayingConfig};
 use crate::eth::AlignedLayerServiceManager;
+use crate::eth::BatchVerifiedEventStream;
 
 mod config;
 mod eth;
@@ -59,6 +60,7 @@ pub struct Batcher {
     last_uploaded_batch_block: Mutex<u64>,
     pre_verification_is_enabled: bool,
     protocol_version: u16,
+    non_paying_config: Option<NonPayingConfig>,
 }
 
 impl Batcher {
@@ -102,6 +104,11 @@ impl Batcher {
         .await
         .expect("Failed to get Batcher Payment Service contract");
 
+        if let Some(non_paying_config) = &config.batcher.non_paying {
+            warn!("Non-paying address configuration detected. Will replace non-paying address {} with configured address {}.",
+                non_paying_config.address, non_paying_config.replacement);
+        }
+
         Self {
             s3_client,
             eth_ws_provider,
@@ -115,6 +122,7 @@ impl Batcher {
             last_uploaded_batch_block: Mutex::new(last_uploaded_batch_block),
             pre_verification_is_enabled: config.batcher.pre_verification_is_enabled,
             protocol_version: PROTOCOL_VERSION,
+            non_paying_config: config.batcher.non_paying,
         }
     }
 
@@ -197,11 +205,18 @@ impl Batcher {
             serde_json::from_str(message.to_text().expect("Message is not text"))
                 .expect("Failed to deserialize task");
 
-        // FIXME: We are not doing anything for the moment with the address from the
-        // sender, this logic should be added for the payment system.
         info!("Verifying message signature...");
         let submitter_addr = if let Ok(addr) = client_msg.verify_signature() {
             info!("Message signature verified");
+
+            let mut addr = addr;
+            if let Some(non_paying_config) = &self.non_paying_config {
+                if addr == non_paying_config.address {
+                    info!("Non-paying address detected. Replacing with configured address");
+                    addr = non_paying_config.replacement;
+                }
+            }
+
             let user_balance = self
                 .payment_service
                 .user_balances(addr)
@@ -215,6 +230,7 @@ impl Batcher {
                     ProtocolError::HandshakeIncomplete,
                 ));
             }
+
             addr
         } else {
             error!("Signature verification error");
@@ -356,10 +372,9 @@ impl Batcher {
         let batch_merkle_tree: MerkleTree<VerificationCommitmentBatch> =
             MerkleTree::build(&batch_data_comm);
 
-        let submitter_addresses: Vec<Address> = finalized_batch
-            .clone()
-            .into_iter()
-            .map(|(_, _, _, addr)| addr)
+        let submitter_addresses = finalized_batch
+            .iter()
+            .map(|(_, _, _, addr)| *addr)
             .collect();
 
         let events = self.service_manager.event::<BatchVerifiedFilter>();
@@ -386,7 +401,10 @@ impl Batcher {
         // connected clients
         let await_batch_verified_fut =
             await_batch_verified_event(&mut stream, &batch_merkle_tree.root);
-        if (timeout(Duration::from_secs(60), await_batch_verified_fut).await).is_err() {
+        if timeout(Duration::from_secs(60), await_batch_verified_fut)
+            .await
+            .is_err()
+        {
             send_timeout_close(finalized_batch).await?;
         } else {
             send_batch_inclusion_data_responses(finalized_batch, &batch_merkle_tree).await;

batcher/aligned/src/main.rs

@@ -1,43 +1,41 @@
-mod errors;
-mod eth;
-mod types;
-
 use std::fs::File;
 use std::io::BufReader;
 use std::io::Write;
 use std::str::FromStr;
 use std::{path::PathBuf, sync::Arc};
 
-use aligned_batcher_lib::types::ClientMessage;
-use aligned_batcher_lib::types::VerificationCommitmentBatch;
-use aligned_batcher_lib::types::VerificationDataCommitment;
+use clap::Subcommand;
+use clap::{Parser, ValueEnum};
 use env_logger::Env;
 use ethers::prelude::*;
+use ethers::utils::hex;
 use futures_util::{
     future,
     stream::{SplitSink, SplitStream},
     SinkExt, StreamExt, TryStreamExt,
 };
 use log::warn;
 use log::{error, info};
+use sha3::{Digest, Keccak256};
 use tokio::{net::TcpStream, sync::Mutex};
 use tokio_tungstenite::connect_async;
 use tokio_tungstenite::tungstenite::Message;
 use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
 
+use aligned_batcher_lib::types::ClientMessage;
+use aligned_batcher_lib::types::VerificationCommitmentBatch;
+use aligned_batcher_lib::types::VerificationDataCommitment;
 use aligned_batcher_lib::types::{BatchInclusionData, ProvingSystemId, VerificationData};
-use clap::Subcommand;
-use ethers::core::rand::thread_rng;
-use ethers::utils::hex;
-use sha3::{Digest, Keccak256};
 
 use crate::errors::BatcherClientError;
 use crate::types::AlignedVerificationData;
 use crate::AlignedCommands::GetVerificationKeyCommitment;
 use crate::AlignedCommands::Submit;
 use crate::AlignedCommands::VerifyProofOnchain;
 
-use clap::{Parser, ValueEnum};
+mod errors;
+mod eth;
+mod types;
 
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
@@ -156,6 +154,8 @@ pub enum ProvingSystemArg {
     Risc0,
 }
 
+const ANVIL_PRIVATE_KEY: &str = "2a871d0798f97d79848a013d4936a73bf4cc922c825d33c1cf7073dff6d409c6"; // Anvil address 9
+
 impl From<ProvingSystemArg> for ProvingSystemId {
     fn from(proving_system: ProvingSystemArg) -> Self {
         match proving_system {
@@ -235,7 +235,7 @@ async fn main() -> Result<(), errors::BatcherClientError> {
                 private_key.parse::<LocalWallet>()?
             } else {
                 warn!("Missing keystore used for payment. This proof will not be included if sent to Eth Mainnet");
-                LocalWallet::new(&mut thread_rng())
+                LocalWallet::from_str(ANVIL_PRIVATE_KEY).expect("Failed to create wallet")
             };
 
             let msg = ClientMessage::new(verification_data, wallet).await;

config-files/config-batcher.yaml

@@ -20,3 +20,7 @@ batcher:
   max_batch_size: 268435456 # 256 MiB
   eth_ws_reconnects: 99999999999999
   pre_verification_is_enabled: true
+  non_paying:
+    address: 0xa0Ee7A142d267C1f36714E4a8F75612F20a79720 # Anvil address 9
+    replacement: 0xf39Fd6e51aad88F6F4ce6aB8827279cffFb92266 # Anvil address 1
+