diff --git a/consensus/types/src/attestation.rs b/consensus/types/src/attestation.rs index de0e86489da..818d1ec2489 100644 --- a/consensus/types/src/attestation.rs +++ b/consensus/types/src/attestation.rs @@ -640,6 +640,28 @@ impl SingleAttestation { }) } } + + pub fn empty_for_signing( + committee_index: u64, + attester_index: u64, + slot: Slot, + beacon_block_root: Hash256, + source: Checkpoint, + target: Checkpoint, + ) -> Self { + Self { + committee_index, + attester_index, + data: AttestationData { + slot, + index: committee_index, + beacon_block_root, + source, + target, + }, + signature: AggregateSignature::infinity(), + } + } } #[cfg(test)] diff --git a/validator_client/lighthouse_validator_store/src/lib.rs b/validator_client/lighthouse_validator_store/src/lib.rs index 67af1d73fed..7c7bc392dba 100644 --- a/validator_client/lighthouse_validator_store/src/lib.rs +++ b/validator_client/lighthouse_validator_store/src/lib.rs @@ -25,6 +25,7 @@ use types::{ SyncCommitteeMessage, SyncSelectionProof, SyncSubnetId, ValidatorRegistrationData, VoluntaryExit, }; +use types::{AggregateSignature, SingleAttestation}; use validator_store::{ DoppelgangerStatus, Error as ValidatorStoreError, ProposalData, SignedBlock, UnsignedBlock, ValidatorStore, @@ -840,6 +841,99 @@ impl ValidatorStore for LighthouseValidatorS } } + async fn sign_single_attestation( + &self, + validator_pubkey: PublicKeyBytes, + single_attestation: &mut SingleAttestation, + current_epoch: Epoch, + ) -> Result<(), Error> { + // Make sure the target epoch is not higher than the current epoch to avoid potential attacks. + if single_attestation.data.target.epoch > current_epoch { + return Err(Error::GreaterThanCurrentEpoch { + epoch: single_attestation.data.target.epoch, + current_epoch, + }); + } + + // Get the signing method and check doppelganger protection. + let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?; + + // Checking for slashing conditions. + let signing_epoch = single_attestation.data.target.epoch; + let signing_context = self.signing_context(Domain::BeaconAttester, signing_epoch); + let domain_hash = signing_context.domain_hash(&self.spec); + + let slashing_status = if signing_method + .requires_local_slashing_protection(self.enable_web3signer_slashing_protection) + { + self.slashing_protection.check_and_insert_attestation( + &validator_pubkey, + &single_attestation.data, + domain_hash, + ) + } else { + Ok(Safe::Valid) + }; + + match slashing_status { + // We can safely sign this attestation. + Ok(Safe::Valid) => { + let signature = signing_method + .get_signature::>( + SignableMessage::AttestationData(&single_attestation.data), + signing_context, + &self.spec, + &self.task_executor, + ) + .await?; + + // Create an aggregate signature from the individual signature + let mut aggregate_signature = AggregateSignature::empty(); + aggregate_signature.add_assign(&signature); + single_attestation.signature = aggregate_signature; + + validator_metrics::inc_counter_vec( + &validator_metrics::SIGNED_ATTESTATIONS_TOTAL, + &[validator_metrics::SUCCESS], + ); + + Ok(()) + } + Ok(Safe::SameData) => { + warn!("Skipping signing of previously signed attestation"); + validator_metrics::inc_counter_vec( + &validator_metrics::SIGNED_ATTESTATIONS_TOTAL, + &[validator_metrics::SAME_DATA], + ); + Err(Error::SameData) + } + Err(NotSafe::UnregisteredValidator(pk)) => { + warn!( + msg = "Carefully consider running with --init-slashing-protection (see --help)", + public_key = format!("{:?}", pk), + "Not signing attestation for unregistered validator" + ); + validator_metrics::inc_counter_vec( + &validator_metrics::SIGNED_ATTESTATIONS_TOTAL, + &[validator_metrics::UNREGISTERED], + ); + Err(Error::Slashable(NotSafe::UnregisteredValidator(pk))) + } + Err(e) => { + crit!( + attestation = format!("{:?}", single_attestation.data), + error = format!("{:?}", e), + "Not signing slashable attestation" + ); + validator_metrics::inc_counter_vec( + &validator_metrics::SIGNED_ATTESTATIONS_TOTAL, + &[validator_metrics::SLASHABLE], + ); + Err(Error::Slashable(e)) + } + } + } + async fn sign_validator_registration_data( &self, validator_registration_data: ValidatorRegistrationData, diff --git a/validator_client/validator_services/src/attestation_service.rs b/validator_client/validator_services/src/attestation_service.rs index e4063cd2117..a935aa72dce 100644 --- a/validator_client/validator_services/src/attestation_service.rs +++ b/validator_client/validator_services/src/attestation_service.rs @@ -10,7 +10,7 @@ use task_executor::TaskExecutor; use tokio::time::{sleep, sleep_until, Duration, Instant}; use tracing::{debug, error, info, trace, warn}; use tree_hash::TreeHash; -use types::{Attestation, AttestationData, ChainSpec, CommitteeIndex, EthSpec, Slot}; +use types::{AttestationData, ChainSpec, CommitteeIndex, EthSpec, SingleAttestation, Slot}; use validator_store::{Error as ValidatorStoreError, ValidatorStore}; /// Builds an `AttestationService`. @@ -379,41 +379,23 @@ impl AttestationService attestation, - Err(err) => { - crit!( - validator = ?duty.pubkey, - ?duty, - ?err, - "Invalid validator duties during signing" - ); - return None; - } - }; + ); + // Sign the SingleAttestation match self .validator_store - .sign_attestation( - duty.pubkey, - duty.validator_committee_index as usize, - &mut attestation, - current_epoch, - ) + .sign_single_attestation(duty.pubkey, &mut single_attestation, current_epoch) .await { - Ok(()) => Some((attestation, duty.validator_index)), + Ok(()) => Some(single_attestation), Err(ValidatorStoreError::UnknownPubkey(pubkey)) => { - // A pubkey can be missing when a validator was recently - // removed via the API. warn!( info = "a validator may have recently been removed from this VC", pubkey = ?pubkey, @@ -438,59 +420,50 @@ impl AttestationService, Vec<_>) = join_all(signing_futures) + let single_attestations: Vec = join_all(signing_futures) .await .into_iter() .flatten() - .unzip(); + .collect(); - if attestations.is_empty() { + if single_attestations.is_empty() { warn!("No attestations were published"); return Ok(None); } + + // Extract validator indices BEFORE moving single_attestations into the closure + let validator_indices: Vec = single_attestations + .iter() + .map(|a| a.attester_index) + .collect(); + let fork_name = self .chain_spec .fork_name_at_slot::(attestation_data.slot); + // Clone single_attestations before using it in the closure + let attestations_to_send = single_attestations.clone(); + // Post the attestations to the BN. match self .beacon_nodes - .request(ApiTopic::Attestations, |beacon_node| async move { - let _timer = validator_metrics::start_timer_vec( - &validator_metrics::ATTESTATION_SERVICE_TIMES, - &[validator_metrics::ATTESTATIONS_HTTP_POST], - ); - - let single_attestations = attestations - .iter() - .zip(validator_indices) - .filter_map(|(a, i)| { - match a.to_single_attestation_with_attester_index(*i) { - Ok(a) => Some(a), - Err(e) => { - // This shouldn't happen unless BN and VC are out of sync with - // respect to the Electra fork. - error!( - error = ?e, - committee_index = attestation_data.index, - slot = slot.as_u64(), - "type" = "unaggregated", - "Unable to convert to SingleAttestation" - ); - None - } - } - }) - .collect::>(); + .request(ApiTopic::Attestations, |beacon_node| { + let attestations = attestations_to_send.clone(); + async move { + let _timer = validator_metrics::start_timer_vec( + &validator_metrics::ATTESTATION_SERVICE_TIMES, + &[validator_metrics::ATTESTATIONS_HTTP_POST], + ); - beacon_node - .post_beacon_pool_attestations_v2::(single_attestations, fork_name) - .await + beacon_node + .post_beacon_pool_attestations_v2::(attestations, fork_name) + .await + } }) .await { Ok(()) => info!( - count = attestations.len(), + count = single_attestations.len(), validator_indices = ?validator_indices, head_block = ?attestation_data.beacon_block_root, committee_index = attestation_data.index, diff --git a/validator_client/validator_store/src/lib.rs b/validator_client/validator_store/src/lib.rs index c3b551c249b..c846f155f37 100644 --- a/validator_client/validator_store/src/lib.rs +++ b/validator_client/validator_store/src/lib.rs @@ -6,8 +6,9 @@ use std::sync::Arc; use types::{ Address, Attestation, AttestationError, BlindedBeaconBlock, Epoch, EthSpec, Graffiti, Hash256, PublicKeyBytes, SelectionProof, Signature, SignedAggregateAndProof, SignedBlindedBeaconBlock, - SignedContributionAndProof, SignedValidatorRegistrationData, Slot, SyncCommitteeContribution, - SyncCommitteeMessage, SyncSelectionProof, SyncSubnetId, ValidatorRegistrationData, + SignedContributionAndProof, SignedValidatorRegistrationData, SingleAttestation, Slot, + SyncCommitteeContribution, SyncCommitteeMessage, SyncSelectionProof, SyncSubnetId, + ValidatorRegistrationData, }; #[derive(Debug, PartialEq, Clone)] @@ -20,6 +21,7 @@ pub enum Error { GreaterThanCurrentSlot { slot: Slot, current_slot: Slot }, GreaterThanCurrentEpoch { epoch: Epoch, current_epoch: Epoch }, UnableToSignAttestation(AttestationError), + ValidatorNotEnabled(PublicKeyBytes), SpecificError(T), } @@ -169,6 +171,14 @@ pub trait ValidatorStore: Send + Sync { /// `ProposalData` fields include defaulting logic described in `get_fee_recipient_defaulting`, /// `get_gas_limit_defaulting`, and `get_builder_proposals_defaulting`. fn proposal_data(&self, pubkey: &PublicKeyBytes) -> Option; + /// Signs a SingleAttestation for a given validator. + /// + fn sign_single_attestation( + &self, + validator_pubkey: PublicKeyBytes, + single_attestation: &mut SingleAttestation, + current_epoch: Epoch, + ) -> impl Future>> + Send; } #[derive(Debug)]