security: data size checks (#219)

Co-authored-by: jotabulacios <jbulacios@fi.uba.ar>

Author: taturosati

batcher/src/config/mod.rs

@@ -10,6 +10,8 @@ pub struct ECDSAConfig {
 pub struct BatcherConfigFromYaml {
     pub block_interval: u64,
     pub batch_size_interval: usize,
+    pub max_proof_size: usize,
+    pub max_batch_size: usize,
 }
 
 #[derive(Debug, Deserialize)]

batcher/src/lib.rs

@@ -36,6 +36,8 @@ pub struct Batcher {
     current_batch: Mutex<Vec<VerificationData>>,
     max_block_interval: u64,
     min_batch_size: usize,
+    max_proof_size: usize,
+    max_batch_size: usize,
     last_uploaded_batch_block: Mutex<u64>,
 }
 
@@ -78,6 +80,8 @@ impl Batcher {
             current_batch: Mutex::new(Vec::new()),
             max_block_interval: config.batcher.block_interval,
             min_batch_size: config.batcher.batch_size_interval,
+            max_proof_size: config.batcher.max_proof_size,
+            max_batch_size: config.batcher.max_batch_size,
             last_uploaded_batch_block: Mutex::new(last_uploaded_batch_block),
         }
     }
@@ -155,7 +159,7 @@ impl Batcher {
             serde_json::from_str(message.to_text().expect("Message is not text"))
                 .expect("Failed to deserialize task");
 
-        if verification_data.verify() {
+        if verification_data.proof.len() <= self.max_proof_size && verification_data.verify() {
             self.add_to_batch(verification_data).await;
         } else {
             // FIXME(marian): Handle this error correctly
@@ -207,14 +211,43 @@ impl Batcher {
 
     async fn process_batch_and_update_state(&self, block_number: u64) -> (Vec<u8>, [u8; 32]) {
         let mut current_batch = self.current_batch.lock().await;
-        let batch_commitment = VerificationCommitmentBatch::from(&(*current_batch));
+
+
+        let mut batch_bytes =
+            serde_json::to_vec(current_batch.as_slice()).expect("Failed to serialize batch");
+
+        let batch_to_send;
+        if batch_bytes.len() > self.max_batch_size {
+            let mut current_batch_end = 0; // not inclusive
+            let mut current_batch_size = 0;
+            for (i, verification_data) in current_batch.iter().enumerate() {
+                let verification_data_bytes = serde_json::to_vec(verification_data)
+                    .expect("Failed to serialize verification data");
+
+                current_batch_size += verification_data_bytes.len();
+                if current_batch_size > self.max_batch_size {
+                    current_batch_end = i;
+                    break;
+                }
+            }
+
+            debug!("Batch size exceeds max batch size, splitting batch at index: {}", current_batch_end);
+            batch_to_send = current_batch.drain(..current_batch_end)
+                .collect::<Vec<_>>();
+
+            info!("# of Elements remaining for next batch: {}", current_batch.len());
+            batch_bytes = serde_json::to_vec(&batch_to_send)
+                .expect("Failed to serialize batch");
+        } else {
+            batch_to_send = current_batch.clone();
+            current_batch.clear();
+        }
+
+        let batch_commitment = VerificationCommitmentBatch::from(&batch_to_send);
         let batch_merkle_tree: MerkleTree<VerificationCommitmentBatch> =
             MerkleTree::build(&batch_commitment.0);
-        let batch_bytes =
-            serde_json::to_vec(current_batch.as_slice()).expect("Failed to serialize batch");
 
-        // update batcher state (clear current batch and update last uploaded batch block)
-        current_batch.clear();
+        // update batcher state (update last uploaded batch block)
         *self.last_uploaded_batch_block.lock().await = block_number;
 
         (batch_bytes, batch_merkle_tree.root)

config-files/config-operator-1.yaml

@@ -27,6 +27,8 @@ operator:
   metadata_url: "https://yetanotherco.github.io/operator_metadata/metadata.json"
   enable_metrics: true
   metrics_ip_port_address: localhost:9092
+  max_batch_size: 268435456 # 256 MiB
+
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"
 private_key_store_path: config-files/devnet/keys/operator-1.ecdsa.key.json

config-files/config-operator-2.yaml

@@ -25,6 +25,8 @@ operator:
   delegation_approver_address: "0x0000000000000000000000000000000000000000"
   staker_opt_out_window_blocks: 0
   metadata_url: "https://yetanotherco.github.io/operator_metadata/metadata.json"
+  max_batch_size: 268435456 # 256 MiB
+
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"
 private_key_store_path: config-files/devnet/keys/operator-2.ecdsa.key.json

config-files/config-operator-3.yaml

@@ -25,6 +25,8 @@ operator:
   delegation_approver_address: "0x0000000000000000000000000000000000000000"
   staker_opt_out_window_blocks: 0
   metadata_url: "https://yetanotherco.github.io/operator_metadata/metadata.json"
+  max_batch_size: 268435456 # 256 MiB
+
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"
 private_key_store_path: config-files/devnet/keys/operator-3.ecdsa.key.json

config-files/config.yaml

@@ -20,7 +20,9 @@ bls:
 ## Batcher configurations
 batcher:
   block_interval: 3
-  batch_size_interval: 5
+  batch_size_interval: 999999999
+  max_proof_size: 67108864 # 64 MiB
+  max_batch_size: 268435456 # 256 MiB
 
 ## Aggregator Configurations
 aggregator:
@@ -40,6 +42,7 @@ operator:
   metadata_url: "https://yetanotherco.github.io/operator_metadata/metadata.json"
   enable_metrics: true
   metrics_ip_port_address: localhost:9092
+  max_batch_size: 268435456 # 256 MiB
 # Operators variables needed for register it in EigenLayer
 el_delegation_manager_address: "0xCf7Ed3AccA5a467e9e704C703E8D87F634fB0Fc9"
 private_key_store_path: config-files/anvil.ecdsa.key.json

core/config/operator.go

@@ -24,6 +24,7 @@ type OperatorConfig struct {
 		RegisterOperatorOnStartup     bool
 		EnableMetrics                 bool
 		MetricsIpPortAddress          string
+		MaxBatchSize                  int64
 	}
 }
 
@@ -38,6 +39,7 @@ type OperatorConfigFromYaml struct {
 		RegisterOperatorOnStartup     bool           `yaml:"register_operator_on_startup"`
 		EnableMetrics                 bool           `yaml:"enable_metrics"`
 		MetricsIpPortAddress          string         `yaml:"metrics_ip_port_address"`
+		MaxBatchSize                  int64          `yaml:"max_batch_size"`
 	} `yaml:"operator"`
 	EcdsaConfigFromYaml EcdsaConfigFromYaml `yaml:"ecdsa"`
 	BlsConfigFromYaml   BlsConfigFromYaml   `yaml:"bls"`
@@ -85,6 +87,7 @@ func NewOperatorConfig(configFilePath string) *OperatorConfig {
 			RegisterOperatorOnStartup     bool
 			EnableMetrics                 bool
 			MetricsIpPortAddress          string
+			MaxBatchSize                  int64
 		}(operatorConfigFromYaml.Operator),
 	}
 }

operator/pkg/operator.go

@@ -133,9 +133,8 @@ func (o *Operator) Start(ctx context.Context) error {
 			// o.Logger.Infof("Received task with index: %d\n", newTaskCreatedLog.TaskIndex)
 			err := o.ProcessNewBatchLog(newBatchLog)
 			if err != nil {
-				o.Logger.Errorf("Proof in batch did not verify", "err", err)
-				// FIXME(marian): This is not how we should handle this error. Just doing this for fast iteration and debug
-				panic("Proof did not verify")
+				o.Logger.Errorf("Batch did not verify", "err", err)
+				continue
 			}
 			responseSignature := o.SignTaskResponse(newBatchLog.BatchMerkleRoot)
 
@@ -162,7 +161,7 @@ func (o *Operator) ProcessNewBatchLog(newBatchLog *servicemanager.ContractAligne
 	verificationDataBatch, err := o.getBatchFromS3(newBatchLog.BatchDataPointer)
 	if err != nil {
 		o.Logger.Errorf("Could not get proofs from S3 bucket: %v", err)
-		return nil
+		return err
 	}
 
 	verificationDataBatchLen := len(verificationDataBatch)

operator/pkg/s3.go

@@ -9,7 +9,22 @@ import (
 
 func (o *Operator) getBatchFromS3(proofUrl string) ([]VerificationData, error) {
 	o.Logger.Infof("Getting batch from S3..., proofUrl: %s", proofUrl)
-	resp, err := http.Get(proofUrl)
+	resp, err := http.Head(proofUrl)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check if the response is OK
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("error getting Proof Head from S3: %s", resp.Status)
+	}
+
+	if resp.ContentLength > o.Config.Operator.MaxBatchSize {
+		return nil, fmt.Errorf("proof size %d exceeds max batch size %d",
+			resp.ContentLength, o.Config.Operator.MaxBatchSize)
+	}
+
+	resp, err = http.Get(proofUrl)
 	if err != nil {
 		return nil, err
 	}