Separate BN for block proposals (#4182)

It is a well-known fact that IP addresses for beacon nodes used by specific validators can be de-anonymized. There is an assumed risk that a malicious user may attempt to DOS validators when producing blocks to prevent chain growth/liveness.

Although there are a number of ideas put forward to address this, there a few simple approaches we can take to mitigate this risk.

Currently, a Lighthouse user is able to set a number of beacon-nodes that their validator client can connect to. If one beacon node is taken offline, it can fallback to another. Different beacon nodes can use VPNs or rotate IPs in order to mask their IPs.

This PR provides an additional setup option which further mitigates attacks of this kind.

This PR introduces a CLI flag --proposer-only to the beacon node. Setting this flag will configure the beacon node to run with minimal peers and crucially will not subscribe to subnets or sync committees. Therefore nodes of this kind should not be identified as nodes connected to validators of any kind.

It also introduces a CLI flag --proposer-nodes to the validator client. Users can then provide a number of beacon nodes (which may or may not run the --proposer-only flag) that the Validator client will use for block production and propagation only. If these nodes fail, the validator client will fallback to the default list of beacon nodes.

Users are then able to set up a number of beacon nodes dedicated to block proposals (which are unlikely to be identified as validator nodes) and point their validator clients to produce blocks on these nodes and attest on other beacon nodes. An attack attempting to prevent liveness on the eth2 network would then need to preemptively find and attack the proposer nodes which is significantly more difficult than the default setup.

This is a follow on from: #3328 

Co-authored-by: Michael Sproul <michael@sigmaprime.io>
Co-authored-by: Paul Hauner <paul@paulhauner.com>

validator_client/src/block_service.rs

@@ -7,8 +7,11 @@ use crate::{
 };
 use crate::{http_metrics::metrics, validator_store::ValidatorStore};
 use environment::RuntimeContext;
+use eth2::BeaconNodeHttpClient;
 use slog::{crit, debug, error, info, trace, warn};
 use slot_clock::SlotClock;
+use std::fmt::Debug;
+use std::future::Future;
 use std::ops::Deref;
 use std::sync::Arc;
 use std::time::Duration;
@@ -45,6 +48,7 @@ pub struct BlockServiceBuilder<T, E: EthSpec> {
     validator_store: Option<Arc<ValidatorStore<T, E>>>,
     slot_clock: Option<Arc<T>>,
     beacon_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
+    proposer_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
     context: Option<RuntimeContext<E>>,
     graffiti: Option<Graffiti>,
     graffiti_file: Option<GraffitiFile>,
@@ -57,6 +61,7 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
             validator_store: None,
             slot_clock: None,
             beacon_nodes: None,
+            proposer_nodes: None,
             context: None,
             graffiti: None,
             graffiti_file: None,
@@ -79,6 +84,11 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
         self
     }
 
+    pub fn proposer_nodes(mut self, proposer_nodes: Arc<BeaconNodeFallback<T, E>>) -> Self {
+        self.proposer_nodes = Some(proposer_nodes);
+        self
+    }
+
     pub fn runtime_context(mut self, context: RuntimeContext<E>) -> Self {
         self.context = Some(context);
         self
@@ -114,6 +124,7 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
                 context: self
                     .context
                     .ok_or("Cannot build BlockService without runtime_context")?,
+                proposer_nodes: self.proposer_nodes,
                 graffiti: self.graffiti,
                 graffiti_file: self.graffiti_file,
                 block_delay: self.block_delay,
@@ -122,11 +133,81 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockServiceBuilder<T, E> {
     }
 }
 
+// Combines a set of non-block-proposing `beacon_nodes` and only-block-proposing
+// `proposer_nodes`.
+pub struct ProposerFallback<T, E: EthSpec> {
+    beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
+    proposer_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
+}
+
+impl<T: SlotClock, E: EthSpec> ProposerFallback<T, E> {
+    // Try `func` on `self.proposer_nodes` first. If that doesn't work, try `self.beacon_nodes`.
+    pub async fn first_success_try_proposers_first<'a, F, O, Err, R>(
+        &'a self,
+        require_synced: RequireSynced,
+        offline_on_failure: OfflineOnFailure,
+        func: F,
+    ) -> Result<O, Errors<Err>>
+    where
+        F: Fn(&'a BeaconNodeHttpClient) -> R + Clone,
+        R: Future<Output = Result<O, Err>>,
+        Err: Debug,
+    {
+        // If there are proposer nodes, try calling `func` on them and return early if they are successful.
+        if let Some(proposer_nodes) = &self.proposer_nodes {
+            if let Ok(result) = proposer_nodes
+                .first_success(require_synced, offline_on_failure, func.clone())
+                .await
+            {
+                return Ok(result);
+            }
+        }
+
+        // If the proposer nodes failed, try on the non-proposer nodes.
+        self.beacon_nodes
+            .first_success(require_synced, offline_on_failure, func)
+            .await
+    }
+
+    // Try `func` on `self.beacon_nodes` first. If that doesn't work, try `self.proposer_nodes`.
+    pub async fn first_success_try_proposers_last<'a, F, O, Err, R>(
+        &'a self,
+        require_synced: RequireSynced,
+        offline_on_failure: OfflineOnFailure,
+        func: F,
+    ) -> Result<O, Errors<Err>>
+    where
+        F: Fn(&'a BeaconNodeHttpClient) -> R + Clone,
+        R: Future<Output = Result<O, Err>>,
+        Err: Debug,
+    {
+        // Try running `func` on the non-proposer beacon nodes.
+        let beacon_nodes_result = self
+            .beacon_nodes
+            .first_success(require_synced, offline_on_failure, func.clone())
+            .await;
+
+        match (beacon_nodes_result, &self.proposer_nodes) {
+            // The non-proposer node call succeed, return the result.
+            (Ok(success), _) => Ok(success),
+            // The non-proposer node call failed, but we don't have any proposer nodes. Return an error.
+            (Err(e), None) => Err(e),
+            // The non-proposer node call failed, try the same call on the proposer nodes.
+            (Err(_), Some(proposer_nodes)) => {
+                proposer_nodes
+                    .first_success(require_synced, offline_on_failure, func)
+                    .await
+            }
+        }
+    }
+}
+
 /// Helper to minimise `Arc` usage.
 pub struct Inner<T, E: EthSpec> {
     validator_store: Arc<ValidatorStore<T, E>>,
     slot_clock: Arc<T>,
     beacon_nodes: Arc<BeaconNodeFallback<T, E>>,
+    proposer_nodes: Option<Arc<BeaconNodeFallback<T, E>>>,
     context: RuntimeContext<E>,
     graffiti: Option<Graffiti>,
     graffiti_file: Option<GraffitiFile>,
@@ -334,16 +415,23 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
         let self_ref = &self;
         let proposer_index = self.validator_store.validator_index(&validator_pubkey);
         let validator_pubkey_ref = &validator_pubkey;
+        let proposer_fallback = ProposerFallback {
+            beacon_nodes: self.beacon_nodes.clone(),
+            proposer_nodes: self.proposer_nodes.clone(),
+        };
 
         info!(
             log,
             "Requesting unsigned block";
             "slot" => slot.as_u64(),
         );
+
         // Request block from first responsive beacon node.
-        let block = self
-            .beacon_nodes
-            .first_success(
+        //
+        // Try the proposer nodes last, since it's likely that they don't have a
+        // great view of attestations on the network.
+        let block = proposer_fallback
+            .first_success_try_proposers_last(
                 RequireSynced::No,
                 OfflineOnFailure::Yes,
                 |beacon_node| async move {
@@ -424,8 +512,12 @@ impl<T: SlotClock + 'static, E: EthSpec> BlockService<T, E> {
         );
 
         // Publish block with first available beacon node.
-        self.beacon_nodes
-            .first_success(
+        //
+        // Try the proposer nodes first, since we've likely gone to efforts to
+        // protect them from DoS attacks and they're most likely to successfully
+        // publish a block.
+        proposer_fallback
+            .first_success_try_proposers_first(
                 RequireSynced::No,
                 OfflineOnFailure::Yes,
                 |beacon_node| async {

validator_client/src/cli.rs

@@ -26,6 +26,15 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 )
                 .takes_value(true),
         )
+        .arg(
+            Arg::with_name("proposer-nodes")
+                .long("proposer-nodes")
+                .value_name("NETWORK_ADDRESSES")
+                .help("Comma-separated addresses to one or more beacon node HTTP APIs. \
+                These specify nodes that are used to send beacon block proposals. A failure will revert back to the standard beacon nodes specified in --beacon-nodes."
+                )
+                .takes_value(true),
+        )
         .arg(
             Arg::with_name("disable-run-on-all")
                 .long("disable-run-on-all")
@@ -118,7 +127,7 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 .value_name("CERTIFICATE-FILES")
                 .takes_value(true)
                 .help("Comma-separated paths to custom TLS certificates to use when connecting \
-                        to a beacon node. These certificates must be in PEM format and are used \
+                        to a beacon node (and/or proposer node). These certificates must be in PEM format and are used \
                         in addition to the OS trust store. Commas must only be used as a \
                         delimiter, and must not be part of the certificate path.")
         )

validator_client/src/config.rs

@@ -29,6 +29,8 @@ pub struct Config {
     ///
     /// Should be similar to `["http://localhost:8080"]`
     pub beacon_nodes: Vec<SensitiveUrl>,
+    /// An optional beacon node used for block proposals only.
+    pub proposer_nodes: Vec<SensitiveUrl>,
     /// If true, the validator client will still poll for duties and produce blocks even if the
     /// beacon node is not synced at startup.
     pub allow_unsynced_beacon_node: bool,
@@ -95,6 +97,7 @@ impl Default for Config {
             validator_dir,
             secrets_dir,
             beacon_nodes,
+            proposer_nodes: Vec::new(),
             allow_unsynced_beacon_node: false,
             disable_auto_discover: false,
             init_slashing_protection: false,
@@ -186,6 +189,14 @@ impl Config {
                 .map_err(|e| format!("Unable to parse beacon node URL: {:?}", e))?];
         }
 
+        if let Some(proposer_nodes) = parse_optional::<String>(cli_args, "proposer_nodes")? {
+            config.proposer_nodes = proposer_nodes
+                .split(',')
+                .map(SensitiveUrl::parse)
+                .collect::<Result<_, _>>()
+                .map_err(|e| format!("Unable to parse proposer node URL: {:?}", e))?;
+        }
+
         if cli_args.is_present("delete-lockfiles") {
             warn!(
                 log,

validator_client/src/lib.rs

@@ -24,6 +24,7 @@ pub use config::Config;
 use initialized_validators::InitializedValidators;
 use lighthouse_metrics::set_gauge;
 use monitoring_api::{MonitoringHttpClient, ProcessType};
+use sensitive_url::SensitiveUrl;
 pub use slashing_protection::{SlashingDatabase, SLASHING_PROTECTION_FILENAME};
 
 use crate::beacon_node_fallback::{
@@ -263,60 +264,70 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
             .checked_sub(1)
             .ok_or_else(|| "No beacon nodes defined.".to_string())?;
 
+        let beacon_node_setup = |x: (usize, &SensitiveUrl)| {
+            let i = x.0;
+            let url = x.1;
+            let slot_duration = Duration::from_secs(context.eth2_config.spec.seconds_per_slot);
+
+            let mut beacon_node_http_client_builder = ClientBuilder::new();
+
+            // Add new custom root certificates if specified.
+            if let Some(certificates) = &config.beacon_nodes_tls_certs {
+                for cert in certificates {
+                    beacon_node_http_client_builder = beacon_node_http_client_builder
+                        .add_root_certificate(load_pem_certificate(cert)?);
+                }
+            }
+
+            let beacon_node_http_client = beacon_node_http_client_builder
+                // Set default timeout to be the full slot duration.
+                .timeout(slot_duration)
+                .build()
+                .map_err(|e| format!("Unable to build HTTP client: {:?}", e))?;
+
+            // Use quicker timeouts if a fallback beacon node exists.
+            let timeouts = if i < last_beacon_node_index && !config.use_long_timeouts {
+                info!(
+                    log,
+                    "Fallback endpoints are available, using optimized timeouts.";
+                );
+                Timeouts {
+                    attestation: slot_duration / HTTP_ATTESTATION_TIMEOUT_QUOTIENT,
+                    attester_duties: slot_duration / HTTP_ATTESTER_DUTIES_TIMEOUT_QUOTIENT,
+                    liveness: slot_duration / HTTP_LIVENESS_TIMEOUT_QUOTIENT,
+                    proposal: slot_duration / HTTP_PROPOSAL_TIMEOUT_QUOTIENT,
+                    proposer_duties: slot_duration / HTTP_PROPOSER_DUTIES_TIMEOUT_QUOTIENT,
+                    sync_committee_contribution: slot_duration
+                        / HTTP_SYNC_COMMITTEE_CONTRIBUTION_TIMEOUT_QUOTIENT,
+                    sync_duties: slot_duration / HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT,
+                    get_beacon_blocks_ssz: slot_duration
+                        / HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT,
+                    get_debug_beacon_states: slot_duration / HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT,
+                    get_deposit_snapshot: slot_duration / HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT,
+                }
+            } else {
+                Timeouts::set_all(slot_duration)
+            };
+
+            Ok(BeaconNodeHttpClient::from_components(
+                url.clone(),
+                beacon_node_http_client,
+                timeouts,
+            ))
+        };
+
         let beacon_nodes: Vec<BeaconNodeHttpClient> = config
             .beacon_nodes
             .iter()
             .enumerate()
-            .map(|(i, url)| {
-                let slot_duration = Duration::from_secs(context.eth2_config.spec.seconds_per_slot);
+            .map(beacon_node_setup)
+            .collect::<Result<Vec<BeaconNodeHttpClient>, String>>()?;
 
-                let mut beacon_node_http_client_builder = ClientBuilder::new();
-
-                // Add new custom root certificates if specified.
-                if let Some(certificates) = &config.beacon_nodes_tls_certs {
-                    for cert in certificates {
-                        beacon_node_http_client_builder = beacon_node_http_client_builder
-                            .add_root_certificate(load_pem_certificate(cert)?);
-                    }
-                }
-
-                let beacon_node_http_client = beacon_node_http_client_builder
-                    // Set default timeout to be the full slot duration.
-                    .timeout(slot_duration)
-                    .build()
-                    .map_err(|e| format!("Unable to build HTTP client: {:?}", e))?;
-
-                // Use quicker timeouts if a fallback beacon node exists.
-                let timeouts = if i < last_beacon_node_index && !config.use_long_timeouts {
-                    info!(
-                        log,
-                        "Fallback endpoints are available, using optimized timeouts.";
-                    );
-                    Timeouts {
-                        attestation: slot_duration / HTTP_ATTESTATION_TIMEOUT_QUOTIENT,
-                        attester_duties: slot_duration / HTTP_ATTESTER_DUTIES_TIMEOUT_QUOTIENT,
-                        liveness: slot_duration / HTTP_LIVENESS_TIMEOUT_QUOTIENT,
-                        proposal: slot_duration / HTTP_PROPOSAL_TIMEOUT_QUOTIENT,
-                        proposer_duties: slot_duration / HTTP_PROPOSER_DUTIES_TIMEOUT_QUOTIENT,
-                        sync_committee_contribution: slot_duration
-                            / HTTP_SYNC_COMMITTEE_CONTRIBUTION_TIMEOUT_QUOTIENT,
-                        sync_duties: slot_duration / HTTP_SYNC_DUTIES_TIMEOUT_QUOTIENT,
-                        get_beacon_blocks_ssz: slot_duration
-                            / HTTP_GET_BEACON_BLOCK_SSZ_TIMEOUT_QUOTIENT,
-                        get_debug_beacon_states: slot_duration
-                            / HTTP_GET_DEBUG_BEACON_STATE_QUOTIENT,
-                        get_deposit_snapshot: slot_duration / HTTP_GET_DEPOSIT_SNAPSHOT_QUOTIENT,
-                    }
-                } else {
-                    Timeouts::set_all(slot_duration)
-                };
-
-                Ok(BeaconNodeHttpClient::from_components(
-                    url.clone(),
-                    beacon_node_http_client,
-                    timeouts,
-                ))
-            })
+        let proposer_nodes: Vec<BeaconNodeHttpClient> = config
+            .proposer_nodes
+            .iter()
+            .enumerate()
+            .map(beacon_node_setup)
             .collect::<Result<Vec<BeaconNodeHttpClient>, String>>()?;
 
         let num_nodes = beacon_nodes.len();
@@ -325,6 +336,12 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
             .map(CandidateBeaconNode::new)
             .collect();
 
+        let proposer_nodes_num = proposer_nodes.len();
+        let proposer_candidates = proposer_nodes
+            .into_iter()
+            .map(CandidateBeaconNode::new)
+            .collect();
+
         // Set the count for beacon node fallbacks excluding the primary beacon node.
         set_gauge(
             &http_metrics::metrics::ETH2_FALLBACK_CONFIGURED,
@@ -349,9 +366,16 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
             log.clone(),
         );
 
+        let mut proposer_nodes: BeaconNodeFallback<_, T> = BeaconNodeFallback::new(
+            proposer_candidates,
+            config.disable_run_on_all,
+            context.eth2_config.spec.clone(),
+            log.clone(),
+        );
+
         // Perform some potentially long-running initialization tasks.
         let (genesis_time, genesis_validators_root) = tokio::select! {
-            tuple = init_from_beacon_node(&beacon_nodes, &context) => tuple?,
+            tuple = init_from_beacon_node(&beacon_nodes, &proposer_nodes, &context) => tuple?,
             () = context.executor.exit() => return Err("Shutting down".to_string())
         };
 
@@ -367,9 +391,14 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
         );
 
         beacon_nodes.set_slot_clock(slot_clock.clone());
+        proposer_nodes.set_slot_clock(slot_clock.clone());
+
         let beacon_nodes = Arc::new(beacon_nodes);
         start_fallback_updater_service(context.clone(), beacon_nodes.clone())?;
 
+        let proposer_nodes = Arc::new(proposer_nodes);
+        start_fallback_updater_service(context.clone(), proposer_nodes.clone())?;
+
         let doppelganger_service = if config.enable_doppelganger_protection {
             Some(Arc::new(DoppelgangerService::new(
                 context
@@ -433,15 +462,21 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
             ctx.shared.write().duties_service = Some(duties_service.clone());
         }
 
-        let block_service = BlockServiceBuilder::new()
+        let mut block_service_builder = BlockServiceBuilder::new()
             .slot_clock(slot_clock.clone())
             .validator_store(validator_store.clone())
             .beacon_nodes(beacon_nodes.clone())
             .runtime_context(context.service_context("block".into()))
             .graffiti(config.graffiti)
             .graffiti_file(config.graffiti_file.clone())
-            .block_delay(config.block_delay)
-            .build()?;
+            .block_delay(config.block_delay);
+
+        // If we have proposer nodes, add them to the block service builder.
+        if proposer_nodes_num > 0 {
+            block_service_builder = block_service_builder.proposer_nodes(proposer_nodes.clone());
+        }
+
+        let block_service = block_service_builder.build()?;
 
         let attestation_service = AttestationServiceBuilder::new()
             .duties_service(duties_service.clone())
@@ -581,13 +616,32 @@ impl<T: EthSpec> ProductionValidatorClient<T> {
 
 async fn init_from_beacon_node<E: EthSpec>(
     beacon_nodes: &BeaconNodeFallback<SystemTimeSlotClock, E>,
+    proposer_nodes: &BeaconNodeFallback<SystemTimeSlotClock, E>,
     context: &RuntimeContext<E>,
 ) -> Result<(u64, Hash256), String> {
     loop {
         beacon_nodes.update_unready_candidates().await;
+        proposer_nodes.update_unready_candidates().await;
+
         let num_available = beacon_nodes.num_available().await;
         let num_total = beacon_nodes.num_total();
-        if num_available > 0 {
+
+        let proposer_available = beacon_nodes.num_available().await;
+        let proposer_total = beacon_nodes.num_total();
+
+        if proposer_total > 0 && proposer_available == 0 {
+            warn!(
+                context.log(),
+                "Unable to connect to a proposer node";
+                "retry in" => format!("{} seconds", RETRY_DELAY.as_secs()),
+                "total_proposers" => proposer_total,
+                "available_proposers" => proposer_available,
+                "total_beacon_nodes" => num_total,
+                "available_beacon_nodes" => num_available,
+            );
+        }
+
+        if num_available > 0 && proposer_available == 0 {
             info!(
                 context.log(),
                 "Initialized beacon node connections";
@@ -595,6 +649,16 @@ async fn init_from_beacon_node<E: EthSpec>(
                 "available" => num_available,
             );
             break;
+        } else if num_available > 0 {
+            info!(
+                context.log(),
+                "Initialized beacon node connections";
+                "total" => num_total,
+                "available" => num_available,
+                "proposers_available" => proposer_available,
+                "proposers_total" => proposer_total,
+            );
+            break;
         } else {
             warn!(
                 context.log(),