Validator blob signing for the unblinded flow (#4096)

* Implement validator blob signing (full block and full blob)

* Fix compilation error and remove redundant slot check

* Fix clippy error

validator_client/src/block_service.rs

@@ -6,9 +6,11 @@ use crate::{
     OfflineOnFailure,
 };
 use crate::{http_metrics::metrics, validator_store::ValidatorStore};
+use bls::SignatureBytes;
 use environment::RuntimeContext;
-use eth2::types::SignedBlockContents;
-use slog::{crit, debug, error, info, trace, warn};
+use eth2::types::{BlockContents, SignedBlockContents};
+use eth2::BeaconNodeHttpClient;
+use slog::{crit, debug, error, info, trace, warn, Logger};
 use slot_clock::SlotClock;
 use std::ops::Deref;
 use std::sync::Arc;
@@ -16,8 +18,8 @@ use std::time::Duration;
 use tokio::sync::mpsc;
 use tokio::time::sleep;
 use types::{
-    AbstractExecPayload, BeaconBlock, BlindedPayload, BlockType, EthSpec, FullPayload, Graffiti,
-    PublicKeyBytes, Slot,
+    AbstractExecPayload, BlindedPayload, BlockType, EthSpec, FullPayload, Graffiti, PublicKeyBytes,
+    Slot,
 };
 
 #[derive(Debug)]
@@ -342,80 +344,46 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
             "slot" => slot.as_u64(),
         );
         // Request block from first responsive beacon node.
-        let block = self
+        let block_contents = self
             .beacon_nodes
             .first_success(
                 RequireSynced::No,
                 OfflineOnFailure::Yes,
-                |beacon_node| async move {
-                    let block: BeaconBlock<E, Payload> = match Payload::block_type() {
-                        BlockType::Full => {
-                            let _get_timer = metrics::start_timer_vec(
-                                &metrics::BLOCK_SERVICE_TIMES,
-                                &[metrics::BEACON_BLOCK_HTTP_GET],
-                            );
-                            beacon_node
-                                .get_validator_blocks::<E, Payload>(
-                                    slot,
-                                    randao_reveal_ref,
-                                    graffiti.as_ref(),
-                                )
-                                .await
-                                .map_err(|e| {
-                                    BlockError::Recoverable(format!(
-                                        "Error from beacon node when producing block: {:?}",
-                                        e
-                                    ))
-                                })?
-                                .data
-                                .into()
-                        }
-                        BlockType::Blinded => {
-                            let _get_timer = metrics::start_timer_vec(
-                                &metrics::BLOCK_SERVICE_TIMES,
-                                &[metrics::BLINDED_BEACON_BLOCK_HTTP_GET],
-                            );
-                            beacon_node
-                                .get_validator_blinded_blocks::<E, Payload>(
-                                    slot,
-                                    randao_reveal_ref,
-                                    graffiti.as_ref(),
-                                )
-                                .await
-                                .map_err(|e| {
-                                    BlockError::Recoverable(format!(
-                                        "Error from beacon node when producing block: {:?}",
-                                        e
-                                    ))
-                                })?
-                                .data
-                        }
-                    };
-
-                    info!(
+                move |beacon_node| {
+                    Self::get_validator_block(
+                        beacon_node,
+                        slot,
+                        randao_reveal_ref,
+                        graffiti,
+                        proposer_index,
                         log,
-                        "Received unsigned block";
-                        "slot" => slot.as_u64(),
-                    );
-                    if proposer_index != Some(block.proposer_index()) {
-                        return Err(BlockError::Recoverable(
-                            "Proposer index does not match block proposer. Beacon chain re-orged"
-                                .to_string(),
-                        ));
-                    }
-
-                    Ok::<_, BlockError>(block)
+                    )
                 },
             )
             .await?;
 
+        let (block, maybe_blob_sidecars) = block_contents.deconstruct();
         let signing_timer = metrics::start_timer(&metrics::BLOCK_SIGNING_TIMES);
-        let signed_block_contents: SignedBlockContents<E, Payload> = self_ref
+
+        let signed_block = self_ref
             .validator_store
             .sign_block::<Payload>(*validator_pubkey_ref, block, current_slot)
             .await
-            .map_err(|e| BlockError::Recoverable(format!("Unable to sign block: {:?}", e)))?
-            .into();
+            .map_err(|e| BlockError::Recoverable(format!("Unable to sign block: {:?}", e)))?;
+
+        let maybe_signed_blobs = match maybe_blob_sidecars {
+            Some(blob_sidecars) => Some(
+                self_ref
+                    .validator_store
+                    .sign_blobs(*validator_pubkey_ref, blob_sidecars)
+                    .await
+                    .map_err(|e| {
+                        BlockError::Recoverable(format!("Unable to sign blob: {:?}", e))
+                    })?,
+            ),
+            None => None,
+        };
+
         let signing_time_ms =
             Duration::from_secs_f64(signing_timer.map_or(0.0, |t| t.stop_and_record())).as_millis();
 
@@ -426,46 +394,19 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
             "signing_time_ms" => signing_time_ms,
         );
 
+        let signed_block_contents = SignedBlockContents::from((signed_block, maybe_signed_blobs));
+
         // Publish block with first available beacon node.
         self.beacon_nodes
             .first_success(
                 RequireSynced::No,
                 OfflineOnFailure::Yes,
                 |beacon_node| async {
-                    match Payload::block_type() {
-                        BlockType::Full => {
-                            let _post_timer = metrics::start_timer_vec(
-                                &metrics::BLOCK_SERVICE_TIMES,
-                                &[metrics::BEACON_BLOCK_HTTP_POST],
-                            );
-                            beacon_node
-                                .post_beacon_blocks(&signed_block_contents)
-                                .await
-                                .map_err(|e| {
-                                    BlockError::Irrecoverable(format!(
-                                        "Error from beacon node when publishing block: {:?}",
-                                        e
-                                    ))
-                                })?
-                        }
-                        BlockType::Blinded => {
-                            let _post_timer = metrics::start_timer_vec(
-                                &metrics::BLOCK_SERVICE_TIMES,
-                                &[metrics::BLINDED_BEACON_BLOCK_HTTP_POST],
-                            );
-                            beacon_node
-                                // TODO: need to be adjusted for blobs
-                                .post_beacon_blinded_blocks(signed_block_contents.signed_block())
-                                .await
-                                .map_err(|e| {
-                                    BlockError::Irrecoverable(format!(
-                                        "Error from beacon node when publishing block: {:?}",
-                                        e
-                                    ))
-                                })?
-                        }
-                    }
-                    Ok::<_, BlockError>(())
+                    Self::publish_signed_block_contents::<Payload>(
+                        &signed_block_contents,
+                        beacon_node,
+                    )
+                    .await
                 },
             )
             .await?;
@@ -482,4 +423,106 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
 
         Ok(())
     }
+
+    async fn publish_signed_block_contents<Payload: AbstractExecPayload<E>>(
+        signed_block_contents: &SignedBlockContents<E, Payload>,
+        beacon_node: &BeaconNodeHttpClient,
+    ) -> Result<(), BlockError> {
+        match Payload::block_type() {
+            BlockType::Full => {
+                let _post_timer = metrics::start_timer_vec(
+                    &metrics::BLOCK_SERVICE_TIMES,
+                    &[metrics::BEACON_BLOCK_HTTP_POST],
+                );
+                beacon_node
+                    .post_beacon_blocks(signed_block_contents)
+                    .await
+                    .map_err(|e| {
+                        BlockError::Irrecoverable(format!(
+                            "Error from beacon node when publishing block: {:?}",
+                            e
+                        ))
+                    })?
+            }
+            BlockType::Blinded => {
+                let _post_timer = metrics::start_timer_vec(
+                    &metrics::BLOCK_SERVICE_TIMES,
+                    &[metrics::BLINDED_BEACON_BLOCK_HTTP_POST],
+                );
+                todo!("need to be adjusted for blobs");
+                // beacon_node
+                //     .post_beacon_blinded_blocks(signed_block_contents.signed_block())
+                //     .await
+                //     .map_err(|e| {
+                //         BlockError::Irrecoverable(format!(
+                //             "Error from beacon node when publishing block: {:?}",
+                //             e
+                //         ))
+                //     })?
+            }
+        }
+        Ok::<_, BlockError>(())
+    }
+
+    async fn get_validator_block<Payload: AbstractExecPayload<E>>(
+        beacon_node: &BeaconNodeHttpClient,
+        slot: Slot,
+        randao_reveal_ref: &SignatureBytes,
+        graffiti: Option<Graffiti>,
+        proposer_index: Option<u64>,
+        log: &Logger,
+    ) -> Result<BlockContents<E, Payload>, BlockError> {
+        let block_contents: BlockContents<E, Payload> = match Payload::block_type() {
+            BlockType::Full => {
+                let _get_timer = metrics::start_timer_vec(
+                    &metrics::BLOCK_SERVICE_TIMES,
+                    &[metrics::BEACON_BLOCK_HTTP_GET],
+                );
+                beacon_node
+                    .get_validator_blocks::<E, Payload>(slot, randao_reveal_ref, graffiti.as_ref())
+                    .await
+                    .map_err(|e| {
+                        BlockError::Recoverable(format!(
+                            "Error from beacon node when producing block: {:?}",
+                            e
+                        ))
+                    })?
+                    .data
+            }
+            BlockType::Blinded => {
+                let _get_timer = metrics::start_timer_vec(
+                    &metrics::BLOCK_SERVICE_TIMES,
+                    &[metrics::BLINDED_BEACON_BLOCK_HTTP_GET],
+                );
+                todo!("implement blinded flow for blobs");
+                // beacon_node
+                //     .get_validator_blinded_blocks::<E, Payload>(
+                //         slot,
+                //         randao_reveal_ref,
+                //         graffiti.as_ref(),
+                //     )
+                //     .await
+                //     .map_err(|e| {
+                //         BlockError::Recoverable(format!(
+                //             "Error from beacon node when producing block: {:?}",
+                //             e
+                //         ))
+                //     })?
+                //     .data
+            }
+        };
+
+        info!(
+            log,
+            "Received unsigned block";
+            "slot" => slot.as_u64(),
+        );
+        if proposer_index != Some(block_contents.block().proposer_index()) {
+            return Err(BlockError::Recoverable(
+                "Proposer index does not match block proposer. Beacon chain re-orged".to_string(),
+            ));
+        }
+
+        Ok::<_, BlockError>(block_contents)
+    }
 }

validator_client/src/signing_method.rs

@@ -37,6 +37,7 @@ pub enum Error {
 pub enum SignableMessage<'a, T: EthSpec, Payload: AbstractExecPayload<T> = FullPayload<T>> {
     RandaoReveal(Epoch),
     BeaconBlock(&'a BeaconBlock<T, Payload>),
+    BlobSidecar(&'a BlobSidecar<T>),
     AttestationData(&'a AttestationData),
     SignedAggregateAndProof(&'a AggregateAndProof<T>),
     SelectionProof(Slot),
@@ -58,6 +59,7 @@ impl<'a, T: EthSpec, Payload: AbstractExecPayload<T>> SignableMessage<'a, T, Pay
         match self {
             SignableMessage::RandaoReveal(epoch) => epoch.signing_root(domain),
             SignableMessage::BeaconBlock(b) => b.signing_root(domain),
+            SignableMessage::BlobSidecar(b) => b.signing_root(domain),
             SignableMessage::AttestationData(a) => a.signing_root(domain),
             SignableMessage::SignedAggregateAndProof(a) => a.signing_root(domain),
             SignableMessage::SelectionProof(slot) => slot.signing_root(domain),
@@ -180,6 +182,10 @@ impl SigningMethod {
                         Web3SignerObject::RandaoReveal { epoch }
                     }
                     SignableMessage::BeaconBlock(block) => Web3SignerObject::beacon_block(block)?,
+                    SignableMessage::BlobSidecar(_) => {
+                        // https://github.com/ConsenSys/web3signer/issues/726
+                        unimplemented!("Web3Signer blob signing not implemented.")
+                    }
                     SignableMessage::AttestationData(a) => Web3SignerObject::Attestation(a),
                     SignableMessage::SignedAggregateAndProof(a) => {
                         Web3SignerObject::AggregateAndProof(a)

validator_client/src/validator_store.rs

@@ -6,6 +6,7 @@ use crate::{
     Config,
 };
 use account_utils::{validator_definitions::ValidatorDefinition, ZeroizeString};
+use eth2::types::VariableList;
 use parking_lot::{Mutex, RwLock};
 use slashing_protection::{
     interchange::Interchange, InterchangeError, NotSafe, Safe, SlashingDatabase,
@@ -19,11 +20,12 @@ use std::sync::Arc;
 use task_executor::TaskExecutor;
 use types::{
     attestation::Error as AttestationError, graffiti::GraffitiString, AbstractExecPayload, Address,
-    AggregateAndProof, Attestation, BeaconBlock, BlindedPayload, ChainSpec, ContributionAndProof,
-    Domain, Epoch, EthSpec, Fork, Graffiti, Hash256, Keypair, PublicKeyBytes, SelectionProof,
-    Signature, SignedAggregateAndProof, SignedBeaconBlock, SignedContributionAndProof, SignedRoot,
-    SignedValidatorRegistrationData, Slot, SyncAggregatorSelectionData, SyncCommitteeContribution,
-    SyncCommitteeMessage, SyncSelectionProof, SyncSubnetId, ValidatorRegistrationData,
+    AggregateAndProof, Attestation, BeaconBlock, BlindedPayload, BlobSidecarList, ChainSpec,
+    ContributionAndProof, Domain, Epoch, EthSpec, Fork, Graffiti, Hash256, Keypair, PublicKeyBytes,
+    SelectionProof, Signature, SignedAggregateAndProof, SignedBeaconBlock, SignedBlobSidecar,
+    SignedBlobSidecarList, SignedContributionAndProof, SignedRoot, SignedValidatorRegistrationData,
+    Slot, SyncAggregatorSelectionData, SyncCommitteeContribution, SyncCommitteeMessage,
+    SyncSelectionProof, SyncSubnetId, ValidatorRegistrationData,
 };
 use validator_dir::ValidatorDir;
 
@@ -531,6 +533,39 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         }
     }
 
+    pub async fn sign_blobs(
+        &self,
+        validator_pubkey: PublicKeyBytes,
+        blob_sidecars: BlobSidecarList<E>,
+    ) -> Result<SignedBlobSidecarList<E>, Error> {
+        let mut signed_blob_sidecars = Vec::new();
+
+        for blob_sidecar in blob_sidecars.into_iter() {
+            let slot = blob_sidecar.slot;
+            let signing_epoch = slot.epoch(E::slots_per_epoch());
+            let signing_context = self.signing_context(Domain::BlobSidecar, signing_epoch);
+            let signing_method = self.doppelganger_checked_signing_method(validator_pubkey)?;
+
+            let signature = signing_method
+                .get_signature::<E, BlindedPayload<E>>(
+                    SignableMessage::BlobSidecar(&blob_sidecar),
+                    signing_context,
+                    &self.spec,
+                    &self.task_executor,
+                )
+                .await?;
+
+            metrics::inc_counter_vec(&metrics::SIGNED_BLOBS_TOTAL, &[metrics::SUCCESS]);
+
+            signed_blob_sidecars.push(SignedBlobSidecar {
+                message: blob_sidecar,
+                signature,
+            });
+        }
+
+        Ok(VariableList::from(signed_blob_sidecars))
+    }
+
     pub async fn sign_attestation(
         &self,
         validator_pubkey: PublicKeyBytes,