Fix duplicate signature aggregation

crates/services/bls_aggregation/src/bls_agg.rs

@@ -316,19 +316,42 @@ impl<A: AvsRegistryService + Send + Sync + Clone + 'static> BlsAggregatorService
             })
             .map_err(|_| BlsAggregationServiceError::TaskExpired)?
         {
+            // check if the operator has already signed for this digest
+            if aggregated_operators
+                .get(&signed_task_digest.task_response_digest)
+                .map(|operators| {
+                    operators
+                        .signers_operator_ids_set
+                        .contains_key(&signed_task_digest.operator_id)
+                })
+                .unwrap_or(false)
+            {
+                signed_task_digest
+                    .signature_verification_channel
+                    .send(Err(SignatureVerificationError::DuplicateSignature))
+                    .await
+                    .map_err(|_| BlsAggregationServiceError::ChannelError)?;
+                continue;
+            }
+
             let verification_result = BlsAggregatorService::<A>::verify_signature(
                 task_index,
                 &signed_task_digest,
                 &operator_state_avs,
             )
             .await;
+            let verification_failed = verification_result.is_err();
 
             signed_task_digest
                 .signature_verification_channel
                 .send(verification_result)
                 .await
                 .map_err(|_| BlsAggregationServiceError::ChannelError)?;
 
+            if verification_failed {
+                continue;
+            }
+
             let operator_state = operator_state_avs
                 .get(&signed_task_digest.operator_id)
                 .unwrap();
@@ -543,7 +566,7 @@ mod tests {
     use alloy_primitives::{B256, U256};
     use eigen_crypto_bls::{BlsG1Point, BlsG2Point, BlsKeyPair, Signature};
     use eigen_services_avsregistry::fake_avs_registry_service::FakeAvsRegistryService;
-    use eigen_types::avs::SignatureVerificationError::IncorrectSignature;
+    use eigen_types::avs::SignatureVerificationError::{DuplicateSignature, IncorrectSignature};
     use eigen_types::operator::{QuorumNum, QuorumThresholdPercentages};
     use eigen_types::{avs::TaskIndex, test::TestOperator};
     use sha2::{Digest, Sha256};
@@ -663,6 +686,113 @@ mod tests {
         assert_eq!(task_index, response.unwrap().unwrap().task_index);
     }
 
+    #[tokio::test]
+    async fn test_1_quorum_2_operator_2_duplicated_signatures() {
+        let test_operator_1 = TestOperator {
+            operator_id: U256::from(1).into(),
+            stake_per_quorum: HashMap::from([(0u8, U256::from(100))]),
+            bls_keypair: BlsKeyPair::new(PRIVATE_KEY_1.into()).unwrap(),
+        };
+        let test_operator_2 = TestOperator {
+            operator_id: U256::from(2).into(),
+            stake_per_quorum: HashMap::from([(0u8, U256::from(100))]),
+            bls_keypair: BlsKeyPair::new(PRIVATE_KEY_2.into()).unwrap(),
+        };
+        let test_operators = vec![test_operator_1.clone(), test_operator_2.clone()];
+        let block_number = 1;
+        let task_index: TaskIndex = 0;
+        let quorum_numbers = vec![0];
+        let quorum_threshold_percentages: QuorumThresholdPercentages = vec![100];
+        let time_to_expiry = Duration::from_secs(1);
+        let task_response = 123; // Initialize with appropriate data
+        let task_response_digest = hash(task_response);
+
+        let fake_avs_registry_service =
+            FakeAvsRegistryService::new(block_number, test_operators.clone());
+        let bls_agg_service = BlsAggregatorService::new(fake_avs_registry_service);
+        bls_agg_service
+            .initialize_new_task(
+                task_index,
+                block_number as u32,
+                quorum_numbers,
+                quorum_threshold_percentages,
+                time_to_expiry,
+            )
+            .await
+            .unwrap();
+
+        let bls_signature_1 = test_operator_1
+            .bls_keypair
+            .sign_message(task_response_digest.as_ref());
+        bls_agg_service
+            .process_new_signature(
+                task_index,
+                task_response_digest,
+                bls_signature_1.clone(),
+                test_operator_1.operator_id,
+            )
+            .await
+            .unwrap();
+
+        let second_signature_processing_result = bls_agg_service
+            .process_new_signature(
+                task_index,
+                task_response_digest,
+                bls_signature_1.clone(),
+                test_operator_1.operator_id,
+            )
+            .await;
+
+        assert_eq!(
+            second_signature_processing_result,
+            Err(BlsAggregationServiceError::SignatureVerificationError(
+                DuplicateSignature
+            ))
+        );
+
+        let bls_signature_2 = test_operator_2
+            .bls_keypair
+            .sign_message(task_response_digest.as_ref());
+        bls_agg_service
+            .process_new_signature(
+                task_index,
+                task_response_digest,
+                bls_signature_2.clone(),
+                test_operator_2.operator_id,
+            )
+            .await
+            .unwrap();
+
+        let quorum_apks_g1 = aggregate_g1_public_keys(&test_operators);
+        let signers_apk_g2 = aggregate_g2_public_keys(&test_operators);
+        let signers_agg_sig_g1 = aggregate_g1_signatures(&[bls_signature_1, bls_signature_2]);
+        let expected_agg_service_response = BlsAggregationServiceResponse {
+            task_index,
+            task_response_digest,
+            non_signers_pub_keys_g1: vec![],
+            quorum_apks_g1: vec![quorum_apks_g1],
+            signers_apk_g2,
+            signers_agg_sig_g1,
+            non_signer_quorum_bitmap_indices: vec![],
+            quorum_apk_indices: vec![],
+            total_stake_indices: vec![],
+            non_signer_stake_indices: vec![],
+        };
+
+        let response = bls_agg_service
+            .aggregated_response_receiver
+            .lock()
+            .await
+            .recv()
+            .await;
+
+        assert_eq!(
+            expected_agg_service_response,
+            response.clone().unwrap().unwrap()
+        );
+        assert_eq!(task_index, response.unwrap().unwrap().task_index);
+    }
+
     #[tokio::test]
     async fn test_1_quorum_3_operator_3_correct_signatures() {
         let test_operator_1 = TestOperator {
@@ -1789,5 +1919,18 @@ mod tests {
             )),
             result
         );
+
+        // Also test that the aggregator service is not affected by the invalid signature, so the task should expire
+        let response = bls_agg_service
+            .aggregated_response_receiver
+            .lock()
+            .await
+            .recv()
+            .await;
+
+        assert_eq!(
+            Err(BlsAggregationServiceError::TaskExpired),
+            response.unwrap()
+        );
     }
 }

crates/types/src/avs.rs

@@ -15,6 +15,8 @@ pub enum SignatureVerificationError {
     OperatorPublicKeyNotFound,
     #[error("operator not found")]
     OperatorNotFound,
+    #[error("duplicate signature error")]
+    DuplicateSignature,
 }
 
 /// Represents a signed task response digest