Register validator api (#3194)

## Issue Addressed

Lays the groundwork for builder API changes by implementing the beacon-API's new `register_validator` endpoint

## Proposed Changes

- Add a routine in the VC that runs on startup (re-try until success), once per epoch or whenever `suggested_fee_recipient` is updated, signing `ValidatorRegistrationData` and sending it to the BN.
  -  TODO: `gas_limit` config options ethereum/builder-specs#17
-  BN only sends VC registration data to builders on demand, but VC registration data *does update* the BN's prepare proposer cache and send an updated fcU to  a local EE. This is necessary for fee recipient consistency between the blinded and full block flow in the event of fallback.  Having the BN only send registration data to builders on demand gives feedback directly to the VC about relay status. Also, since the BN has no ability to sign these messages anyways (so couldn't refresh them if it wanted), and validator registration is independent of the BN head, I think this approach makes sense. 
- Adds upcoming consensus spec changes for this PR ethereum/consensus-specs#2884
  -  I initially applied the bit mask based on a configured application domain.. but I ended up just hard coding it here instead because that's how it's spec'd in the builder repo. 
  -  Should application mask appear in the api?



Co-authored-by: realbigsean <sean@sigmaprime.io>

beacon_node/http_api/Cargo.toml

@@ -39,6 +39,7 @@ environment = { path = "../../lighthouse/environment" }
 tree_hash = "0.4.1"
 sensitive_url = { path = "../../common/sensitive_url" }
 logging = { path = "../../common/logging" }
+serde_json = "1.0.58"
 
 [[test]]
 name = "bn_http_api_tests"

beacon_node/http_api/src/lib.rs

@@ -49,8 +49,8 @@ use types::{
     BlindedPayload, CommitteeCache, ConfigAndPreset, Epoch, EthSpec, ForkName, FullPayload,
     ProposerPreparationData, ProposerSlashing, RelativeEpoch, Signature, SignedAggregateAndProof,
     SignedBeaconBlock, SignedBeaconBlockMerge, SignedBlindedBeaconBlock,
-    SignedContributionAndProof, SignedVoluntaryExit, Slot, SyncCommitteeMessage,
-    SyncContributionData,
+    SignedContributionAndProof, SignedValidatorRegistrationData, SignedVoluntaryExit, Slot,
+    SyncCommitteeMessage, SyncContributionData,
 };
 use version::{
     add_consensus_version_header, fork_versioned_response, inconsistent_fork_rejection,
@@ -2408,12 +2408,10 @@ pub fn serve<T: BeaconChainTypes>(
         .and(warp::path::end())
         .and(not_while_syncing_filter.clone())
         .and(chain_filter.clone())
-        .and(warp::addr::remote())
         .and(log_filter.clone())
         .and(warp::body::json())
         .and_then(
             |chain: Arc<BeaconChain<T>>,
-             client_addr: Option<SocketAddr>,
              log: Logger,
              preparation_data: Vec<ProposerPreparationData>| {
                 blocking_json_task(move || {
@@ -2430,9 +2428,6 @@ pub fn serve<T: BeaconChainTypes>(
                         log,
                         "Received proposer preparation data";
                         "count" => preparation_data.len(),
-                        "client" => client_addr
-                            .map(|a| a.to_string())
-                            .unwrap_or_else(|| "unknown".to_string()),
                     );
 
                     execution_layer
@@ -2455,6 +2450,82 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // POST validator/register_validator
+    let post_validator_register_validator = eth1_v1
+        .and(warp::path("validator"))
+        .and(warp::path("register_validator"))
+        .and(warp::path::end())
+        .and(chain_filter.clone())
+        .and(log_filter.clone())
+        .and(warp::body::json())
+        .and_then(
+            |chain: Arc<BeaconChain<T>>,
+             log: Logger,
+             register_val_data: Vec<SignedValidatorRegistrationData>| {
+                blocking_json_task(move || {
+                    let execution_layer = chain
+                        .execution_layer
+                        .as_ref()
+                        .ok_or(BeaconChainError::ExecutionLayerMissing)
+                        .map_err(warp_utils::reject::beacon_chain_error)?;
+                    let current_epoch = chain
+                        .slot_clock
+                        .now_or_genesis()
+                        .ok_or(BeaconChainError::UnableToReadSlot)
+                        .map_err(warp_utils::reject::beacon_chain_error)?
+                        .epoch(T::EthSpec::slots_per_epoch());
+
+                    debug!(
+                        log,
+                        "Received register validator request";
+                        "count" => register_val_data.len(),
+                    );
+
+                    let preparation_data = register_val_data
+                        .iter()
+                        .filter_map(|register_data| {
+                            chain
+                                .validator_index(&register_data.message.pubkey)
+                                .ok()
+                                .flatten()
+                                .map(|validator_index| ProposerPreparationData {
+                                    validator_index: validator_index as u64,
+                                    fee_recipient: register_data.message.fee_recipient,
+                                })
+                        })
+                        .collect::<Vec<_>>();
+
+                    debug!(
+                        log,
+                        "Resolved validator request pubkeys";
+                        "count" => preparation_data.len()
+                    );
+
+                    // Update the prepare beacon proposer cache based on this request.
+                    execution_layer
+                        .update_proposer_preparation_blocking(current_epoch, &preparation_data)
+                        .map_err(|_e| {
+                            warp_utils::reject::custom_bad_request(
+                                "error processing proposer preparations".to_string(),
+                            )
+                        })?;
+
+                    // Call prepare beacon proposer blocking with the latest update in order to make
+                    // sure we have a local payload to fall back to in the event of the blined block
+                    // flow failing.
+                    chain.prepare_beacon_proposer_blocking().map_err(|e| {
+                        warp_utils::reject::custom_bad_request(format!(
+                            "error updating proposer preparations: {:?}",
+                            e
+                        ))
+                    })?;
+
+                    //TODO(sean): In the MEV-boost PR, add a call here to send the update request to the builder
+
+                    Ok(())
+                })
+            },
+        );
     // POST validator/sync_committee_subscriptions
     let post_validator_sync_committee_subscriptions = eth1_v1
         .and(warp::path("validator"))
@@ -3008,6 +3079,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(post_validator_beacon_committee_subscriptions.boxed())
                 .or(post_validator_sync_committee_subscriptions.boxed())
                 .or(post_validator_prepare_beacon_proposer.boxed())
+                .or(post_validator_register_validator.boxed())
                 .or(post_lighthouse_liveness.boxed())
                 .or(post_lighthouse_database_reconstruct.boxed())
                 .or(post_lighthouse_database_historical_blocks.boxed())

beacon_node/http_api/tests/tests.rs

@@ -11,6 +11,7 @@ use eth2::{
     types::*,
     BeaconNodeHttpClient, Error, StatusCode, Timeouts,
 };
+use execution_layer::test_utils::MockExecutionLayer;
 use futures::stream::{Stream, StreamExt};
 use futures::FutureExt;
 use lighthouse_network::{Enr, EnrExt, PeerId};
@@ -24,6 +25,7 @@ use task_executor::test_utils::TestRuntime;
 use tokio::sync::{mpsc, oneshot};
 use tokio::time::Duration;
 use tree_hash::TreeHash;
+use types::application_domain::ApplicationDomain;
 use types::{
     AggregateSignature, BeaconState, BitList, Domain, EthSpec, Hash256, Keypair, MainnetEthSpec,
     RelativeEpoch, SelectionProof, SignedRoot, Slot,
@@ -64,6 +66,9 @@ struct ApiTester {
     network_rx: mpsc::UnboundedReceiver<NetworkMessage<E>>,
     local_enr: Enr,
     external_peer_id: PeerId,
+    // This is never directly accessed, but adding it creates a payload cache, which we use in tests here.
+    #[allow(dead_code)]
+    mock_el: Option<MockExecutionLayer<E>>,
     _runtime: TestRuntime,
 }
 
@@ -80,6 +85,7 @@ impl ApiTester {
             .spec(spec.clone())
             .deterministic_keypairs(VALIDATOR_COUNT)
             .fresh_ephemeral_store()
+            .mock_execution_layer()
             .build();
 
         harness.advance_slot();
@@ -214,6 +220,7 @@ impl ApiTester {
             network_rx,
             local_enr,
             external_peer_id,
+            mock_el: harness.mock_execution_layer,
             _runtime: harness.runtime,
         }
     }
@@ -293,6 +300,7 @@ impl ApiTester {
             network_rx,
             local_enr,
             external_peer_id,
+            mock_el: None,
             _runtime: harness.runtime,
         }
     }
@@ -2226,6 +2234,66 @@ impl ApiTester {
         self
     }
 
+    pub async fn test_post_validator_register_validator(self) -> Self {
+        let mut registrations = vec![];
+        let mut fee_recipients = vec![];
+
+        let fork = self.chain.head().unwrap().beacon_state.fork();
+
+        for (val_index, keypair) in self.validator_keypairs.iter().enumerate() {
+            let pubkey = keypair.pk.compress();
+            let fee_recipient = Address::from_low_u64_be(val_index as u64);
+
+            let data = ValidatorRegistrationData {
+                fee_recipient,
+                gas_limit: 0,
+                timestamp: 0,
+                pubkey,
+            };
+            let domain = self.chain.spec.get_domain(
+                Epoch::new(0),
+                Domain::ApplicationMask(ApplicationDomain::Builder),
+                &fork,
+                Hash256::zero(),
+            );
+            let message = data.signing_root(domain);
+            let signature = keypair.sk.sign(message);
+
+            fee_recipients.push(fee_recipient);
+            registrations.push(SignedValidatorRegistrationData {
+                message: data,
+                signature,
+            });
+        }
+
+        self.client
+            .post_validator_register_validator(&registrations)
+            .await
+            .unwrap();
+
+        for (val_index, (_, fee_recipient)) in self
+            .chain
+            .head()
+            .unwrap()
+            .beacon_state
+            .validators()
+            .into_iter()
+            .zip(fee_recipients.into_iter())
+            .enumerate()
+        {
+            let actual = self
+                .chain
+                .execution_layer
+                .as_ref()
+                .unwrap()
+                .get_suggested_fee_recipient(val_index as u64)
+                .await;
+            assert_eq!(actual, fee_recipient);
+        }
+
+        self
+    }
+
     #[cfg(target_os = "linux")]
     pub async fn test_get_lighthouse_health(self) -> Self {
         self.client.get_lighthouse_health().await.unwrap();
@@ -2973,6 +3041,14 @@ async fn get_validator_beacon_committee_subscriptions() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn post_validator_register_validator() {
+    ApiTester::new()
+        .await
+        .test_post_validator_register_validator()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn lighthouse_endpoints() {
     ApiTester::new()

book/src/api-vc-endpoints.md

@@ -134,6 +134,7 @@ Typical Responses | 200
         "DOMAIN_VOLUNTARY_EXIT": "0x04000000",
         "DOMAIN_SELECTION_PROOF": "0x05000000",
         "DOMAIN_AGGREGATE_AND_PROOF": "0x06000000",
+        "DOMAIN_APPLICATION_MASK": "0x00000001",
         "MAX_VALIDATORS_PER_COMMITTEE": "2048",
         "SLOTS_PER_EPOCH": "32",
         "EPOCHS_PER_ETH1_VOTING_PERIOD": "32",

common/eth2/src/lib.rs

@@ -929,6 +929,23 @@ impl BeaconNodeHttpClient {
         Ok(())
     }
 
+    /// `POST validator/register_validator`
+    pub async fn post_validator_register_validator(
+        &self,
+        registration_data: &[SignedValidatorRegistrationData],
+    ) -> Result<(), Error> {
+        let mut path = self.eth_path(V1)?;
+
+        path.path_segments_mut()
+            .map_err(|()| Error::InvalidUrl(self.server.clone()))?
+            .push("validator")
+            .push("register_validator");
+
+        self.post(path, &registration_data).await?;
+
+        Ok(())
+    }
+
     /// `GET config/fork_schedule`
     pub async fn get_config_fork_schedule(&self) -> Result<GenericResponse<Vec<Fork>>, Error> {
         let mut path = self.eth_path(V1)?;

consensus/types/src/application_domain.rs

@@ -0,0 +1,16 @@
+/// This value is an application index of 0 with the bitmask applied (so it's equivalent to the bit mask).
+/// Little endian hex: 0x00000001, Binary: 1000000000000000000000000
+pub const APPLICATION_DOMAIN_BUILDER: u32 = 16777216;
+
+#[derive(Debug, PartialEq, Clone, Copy)]
+pub enum ApplicationDomain {
+    Builder,
+}
+
+impl ApplicationDomain {
+    pub fn get_domain_constant(&self) -> u32 {
+        match self {
+            ApplicationDomain::Builder => APPLICATION_DOMAIN_BUILDER,
+        }
+    }
+}