Doppelganger detection (#2230)

## Issue Addressed

Resolves #2069 

## Proposed Changes

- Adds a `--doppelganger-detection` flag
- Adds a `lighthouse/seen_validators` endpoint, which will make it so the lighthouse VC is not interopable with other client beacon nodes if the `--doppelganger-detection` flag is used, but hopefully this will become standardized. Relevant Eth2 API repo issue: https://github.com/ethereum/eth2.0-APIs/issues/64
- If the `--doppelganger-detection` flag is used, the VC will wait until the beacon node is synced, and then wait an additional 2 epochs. The reason for this is to make sure the beacon node is able to subscribe to the subnets our validators should be attesting on. I think an alternative would be to have the beacon node subscribe to all subnets for 2+ epochs on startup by default.

## Additional Info

I'd like to add tests and would appreciate feedback. 

TODO:  handle validators started via the API, potentially make this default behavior

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

validator_client/src/validator_store.rs

@@ -1,21 +1,36 @@
 use crate::{
-    fork_service::ForkService, http_metrics::metrics, initialized_validators::InitializedValidators,
+    doppelganger_service::DoppelgangerService, fork_service::ForkService, http_metrics::metrics,
+    initialized_validators::InitializedValidators,
 };
 use account_utils::{validator_definitions::ValidatorDefinition, ZeroizeString};
 use parking_lot::{Mutex, RwLock};
 use slashing_protection::{NotSafe, Safe, SlashingDatabase};
 use slog::{crit, error, info, warn, Logger};
 use slot_clock::SlotClock;
+use std::iter::FromIterator;
 use std::path::Path;
 use std::sync::Arc;
-use tempfile::TempDir;
 use types::{
-    graffiti::GraffitiString, Attestation, BeaconBlock, ChainSpec, Domain, Epoch, EthSpec, Fork,
-    Graffiti, Hash256, Keypair, PublicKeyBytes, SelectionProof, Signature, SignedAggregateAndProof,
-    SignedBeaconBlock, SignedRoot, Slot,
+    attestation::Error as AttestationError, graffiti::GraffitiString, Attestation, BeaconBlock,
+    ChainSpec, Domain, Epoch, EthSpec, Fork, Graffiti, Hash256, Keypair, PublicKeyBytes,
+    SelectionProof, Signature, SignedAggregateAndProof, SignedBeaconBlock, SignedRoot, Slot,
 };
 use validator_dir::ValidatorDir;
 
+pub use crate::doppelganger_service::DoppelgangerStatus;
+
+#[derive(Debug, PartialEq)]
+pub enum Error {
+    DoppelgangerProtected(PublicKeyBytes),
+    UnknownToDoppelgangerService(PublicKeyBytes),
+    UnknownPubkey(PublicKeyBytes),
+    Slashable(NotSafe),
+    SameData,
+    GreaterThanCurrentSlot { slot: Slot, current_slot: Slot },
+    GreaterThanCurrentEpoch { epoch: Epoch, current_epoch: Epoch },
+    UnableToSignAttestation(AttestationError),
+}
+
 /// Number of epochs of slashing protection history to keep.
 ///
 /// This acts as a maximum safe-guard against clock drift.
@@ -46,7 +61,6 @@ impl PartialEq for LocalValidator {
     }
 }
 
-#[derive(Clone)]
 pub struct ValidatorStore<T, E: EthSpec> {
     validators: Arc<RwLock<InitializedValidators>>,
     slashing_protection: SlashingDatabase,
@@ -54,8 +68,9 @@ pub struct ValidatorStore<T, E: EthSpec> {
     genesis_validators_root: Hash256,
     spec: Arc<ChainSpec>,
     log: Logger,
-    temp_dir: Option<Arc<TempDir>>,
+    doppelganger_service: Option<Arc<DoppelgangerService>>,
     fork_service: ForkService<T, E>,
+    slot_clock: T,
 }
 
 impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
@@ -65,6 +80,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         genesis_validators_root: Hash256,
         spec: ChainSpec,
         fork_service: ForkService<T, E>,
+        doppelganger_service: Option<Arc<DoppelgangerService>>,
         log: Logger,
     ) -> Self {
         Self {
@@ -73,12 +89,32 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
             slashing_protection_last_prune: Arc::new(Mutex::new(Epoch::new(0))),
             genesis_validators_root,
             spec: Arc::new(spec),
-            log,
-            temp_dir: None,
+            log: log.clone(),
+            doppelganger_service,
+            slot_clock: fork_service.slot_clock(),
             fork_service,
         }
     }
 
+    /// Register all local validators in doppelganger protection to try and prevent instances of
+    /// duplicate validators operating on the network at the same time.
+    ///
+    /// This function has no effect if doppelganger protection is disabled.
+    pub fn register_all_in_doppelganger_protection_if_enabled(&self) -> Result<(), String> {
+        if let Some(doppelganger_service) = &self.doppelganger_service {
+            for pubkey in self.validators.read().iter_voting_pubkeys() {
+                doppelganger_service.register_new_validator::<E, _>(*pubkey, &self.slot_clock)?
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Returns `true` if doppelganger protection is enabled, or else `false`.
+    pub fn doppelganger_protection_enabled(&self) -> bool {
+        self.doppelganger_service.is_some()
+    }
+
     pub fn initialized_validators(&self) -> Arc<RwLock<InitializedValidators>> {
         self.validators.clone()
     }
@@ -105,12 +141,19 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         )
         .map_err(|e| format!("failed to create validator definitions: {:?}", e))?;
 
+        let validator_pubkey = validator_def.voting_public_key.compress();
+
         self.slashing_protection
-            .register_validator(validator_def.voting_public_key.compress())
+            .register_validator(validator_pubkey)
             .map_err(|e| format!("failed to register validator: {:?}", e))?;
 
         validator_def.enabled = enable;
 
+        if let Some(doppelganger_service) = &self.doppelganger_service {
+            doppelganger_service
+                .register_new_validator::<E, _>(validator_pubkey, &self.slot_clock)?;
+        }
+
         self.validators
             .write()
             .add_definition(validator_def.clone())
@@ -120,14 +163,92 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         Ok(validator_def)
     }
 
-    pub fn voting_pubkeys(&self) -> Vec<PublicKeyBytes> {
-        self.validators
+    /// Attempts to resolve the pubkey to a validator index.
+    ///
+    /// It may return `None` if the `pubkey` is:
+    ///
+    /// - Unknown.
+    /// - Known, but with an unknown index.
+    pub fn validator_index(&self, pubkey: &PublicKeyBytes) -> Option<u64> {
+        self.validators.read().get_index(pubkey)
+    }
+
+    /// Returns all voting pubkeys for all enabled validators.
+    ///
+    /// The `filter_func` allows for filtering pubkeys based upon their `DoppelgangerStatus`. There
+    /// are two primary functions used here:
+    ///
+    /// - `DoppelgangerStatus::only_safe`: only returns pubkeys which have passed doppelganger
+    ///     protection and are safe-enough to sign messages.
+    /// - `DoppelgangerStatus::ignored`: returns all the pubkeys from `only_safe` *plus* those still
+    ///     undergoing protection. This is useful for collecting duties or other non-signing tasks.
+    #[allow(clippy::needless_collect)] // Collect is required to avoid holding a lock.
+    pub fn voting_pubkeys<I, F>(&self, filter_func: F) -> I
+    where
+        I: FromIterator<PublicKeyBytes>,
+        F: Fn(DoppelgangerStatus) -> Option<PublicKeyBytes>,
+    {
+        // Collect all the pubkeys first to avoid interleaving locks on `self.validators` and
+        // `self.doppelganger_service()`.
+        let pubkeys = self
+            .validators
             .read()
             .iter_voting_pubkeys()
             .cloned()
+            .collect::<Vec<_>>();
+
+        pubkeys
+            .into_iter()
+            .map(|pubkey| {
+                self.doppelganger_service
+                    .as_ref()
+                    .map(|doppelganger_service| doppelganger_service.validator_status(pubkey))
+                    // Allow signing on all pubkeys if doppelganger protection is disabled.
+                    .unwrap_or_else(|| DoppelgangerStatus::SigningEnabled(pubkey))
+            })
+            .filter_map(filter_func)
             .collect()
     }
 
+    /// Returns doppelganger statuses for all enabled validators.
+    #[allow(clippy::needless_collect)] // Collect is required to avoid holding a lock.
+    pub fn doppelganger_statuses(&self) -> Vec<DoppelgangerStatus> {
+        // Collect all the pubkeys first to avoid interleaving locks on `self.validators` and
+        // `self.doppelganger_service`.
+        let pubkeys = self
+            .validators
+            .read()
+            .iter_voting_pubkeys()
+            .cloned()
+            .collect::<Vec<_>>();
+
+        pubkeys
+            .into_iter()
+            .map(|pubkey| {
+                self.doppelganger_service
+                    .as_ref()
+                    .map(|doppelganger_service| doppelganger_service.validator_status(pubkey))
+                    // Allow signing on all pubkeys if doppelganger protection is disabled.
+                    .unwrap_or_else(|| DoppelgangerStatus::SigningEnabled(pubkey))
+            })
+            .collect()
+    }
+
+    /// Check if the `validator_pubkey` is permitted by the doppleganger protection to sign
+    /// messages.
+    pub fn doppelganger_protection_allows_signing(&self, validator_pubkey: PublicKeyBytes) -> bool {
+        self.doppelganger_service
+            .as_ref()
+            // If there's no doppelganger service then we assume it is purposefully disabled and
+            // declare that all keys are safe with regard to it.
+            .map_or(true, |doppelganger_service| {
+                doppelganger_service
+                    .validator_status(validator_pubkey)
+                    .only_safe()
+                    .is_some()
+            })
+    }
+
     pub fn num_voting_validators(&self) -> usize {
         self.validators.read().num_enabled()
     }
@@ -136,25 +257,56 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         self.fork_service.fork()
     }
 
+    /// Runs `func`, providing it access to the `Keypair` corresponding to `validator_pubkey`.
+    ///
+    /// This forms the canonical point for accessing the secret key of some validator. It is
+    /// structured as a `with_...` function since we need to pass-through a read-lock in order to
+    /// access the keypair.
+    ///
+    /// Access to keypairs might be restricted by other internal mechanisms (e.g., doppleganger
+    /// protection).
+    ///
+    /// ## Warning
+    ///
+    /// This function takes a read-lock on `self.validators`. To prevent deadlocks, it is advised to
+    /// never take any sort of concurrency lock inside this function.
+    fn with_validator_keypair<F, R>(
+        &self,
+        validator_pubkey: PublicKeyBytes,
+        func: F,
+    ) -> Result<R, Error>
+    where
+        F: FnOnce(&Keypair) -> R,
+    {
+        // If the doppelganger service is active, check to ensure it explicitly permits signing by
+        // this validator.
+        if !self.doppelganger_protection_allows_signing(validator_pubkey) {
+            return Err(Error::DoppelgangerProtected(validator_pubkey));
+        }
+
+        let validators_lock = self.validators.read();
+
+        Ok(func(
+            validators_lock
+                .voting_keypair(&validator_pubkey)
+                .ok_or(Error::UnknownPubkey(validator_pubkey))?,
+        ))
+    }
+
     pub fn randao_reveal(
         &self,
-        validator_pubkey: &PublicKeyBytes,
+        validator_pubkey: PublicKeyBytes,
         epoch: Epoch,
-    ) -> Option<Signature> {
-        self.validators
-            .read()
-            .voting_keypair(validator_pubkey)
-            .map(|voting_keypair| {
-                let domain = self.spec.get_domain(
-                    epoch,
-                    Domain::Randao,
-                    &self.fork(),
-                    self.genesis_validators_root,
-                );
-                let message = epoch.signing_root(domain);
+    ) -> Result<Signature, Error> {
+        let domain = self.spec.get_domain(
+            epoch,
+            Domain::Randao,
+            &self.fork(),
+            self.genesis_validators_root,
+        );
+        let message = epoch.signing_root(domain);
 
-                voting_keypair.sk.sign(message)
-            })
+        self.with_validator_keypair(validator_pubkey, |keypair| keypair.sk.sign(message))
     }
 
     pub fn graffiti(&self, validator_pubkey: &PublicKeyBytes) -> Option<Graffiti> {
@@ -163,10 +315,10 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
 
     pub fn sign_block(
         &self,
-        validator_pubkey: &PublicKeyBytes,
+        validator_pubkey: PublicKeyBytes,
         block: BeaconBlock<E>,
         current_slot: Slot,
-    ) -> Option<SignedBeaconBlock<E>> {
+    ) -> Result<SignedBeaconBlock<E>, Error> {
         // Make sure the block slot is not higher than the current slot to avoid potential attacks.
         if block.slot() > current_slot {
             warn!(
@@ -175,7 +327,10 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                 "block_slot" => block.slot().as_u64(),
                 "current_slot" => current_slot.as_u64()
             );
-            return None;
+            return Err(Error::GreaterThanCurrentSlot {
+                slot: block.slot(),
+                current_slot,
+            });
         }
 
         // Check for slashing conditions.
@@ -188,25 +343,19 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         );
 
         let slashing_status = self.slashing_protection.check_and_insert_block_proposal(
-            validator_pubkey,
+            &validator_pubkey,
             &block.block_header(),
             domain,
         );
 
         match slashing_status {
-            // We can safely sign this block.
+            // We can safely sign this block without slashing.
             Ok(Safe::Valid) => {
-                let validators = self.validators.read();
-                let voting_keypair = validators.voting_keypair(validator_pubkey)?;
-
                 metrics::inc_counter_vec(&metrics::SIGNED_BLOCKS_TOTAL, &[metrics::SUCCESS]);
 
-                Some(block.sign(
-                    &voting_keypair.sk,
-                    &fork,
-                    self.genesis_validators_root,
-                    &self.spec,
-                ))
+                self.with_validator_keypair(validator_pubkey, move |keypair| {
+                    block.sign(&keypair.sk, &fork, self.genesis_validators_root, &self.spec)
+                })
             }
             Ok(Safe::SameData) => {
                 warn!(
@@ -214,7 +363,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                     "Skipping signing of previously signed block";
                 );
                 metrics::inc_counter_vec(&metrics::SIGNED_BLOCKS_TOTAL, &[metrics::SAME_DATA]);
-                None
+                Err(Error::SameData)
             }
             Err(NotSafe::UnregisteredValidator(pk)) => {
                 warn!(
@@ -224,7 +373,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                     "public_key" => format!("{:?}", pk)
                 );
                 metrics::inc_counter_vec(&metrics::SIGNED_BLOCKS_TOTAL, &[metrics::UNREGISTERED]);
-                None
+                Err(Error::Slashable(NotSafe::UnregisteredValidator(pk)))
             }
             Err(e) => {
                 crit!(
@@ -233,21 +382,24 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                     "error" => format!("{:?}", e)
                 );
                 metrics::inc_counter_vec(&metrics::SIGNED_BLOCKS_TOTAL, &[metrics::SLASHABLE]);
-                None
+                Err(Error::Slashable(e))
             }
         }
     }
 
     pub fn sign_attestation(
         &self,
-        validator_pubkey: &PublicKeyBytes,
+        validator_pubkey: PublicKeyBytes,
         validator_committee_position: usize,
         attestation: &mut Attestation<E>,
         current_epoch: Epoch,
-    ) -> Option<()> {
+    ) -> Result<(), Error> {
         // Make sure the target epoch is not higher than the current epoch to avoid potential attacks.
         if attestation.data.target.epoch > current_epoch {
-            return None;
+            return Err(Error::GreaterThanCurrentEpoch {
+                epoch: attestation.data.target.epoch,
+                current_epoch,
+            });
         }
 
         // Checking for slashing conditions.
@@ -260,7 +412,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
             self.genesis_validators_root,
         );
         let slashing_status = self.slashing_protection.check_and_insert_attestation(
-            validator_pubkey,
+            &validator_pubkey,
             &attestation.data,
             domain,
         );
@@ -268,29 +420,20 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         match slashing_status {
             // We can safely sign this attestation.
             Ok(Safe::Valid) => {
-                let validators = self.validators.read();
-                let voting_keypair = validators.voting_keypair(validator_pubkey)?;
-
-                attestation
-                    .sign(
-                        &voting_keypair.sk,
+                self.with_validator_keypair(validator_pubkey, |keypair| {
+                    attestation.sign(
+                        &keypair.sk,
                         validator_committee_position,
                         &fork,
                         self.genesis_validators_root,
                         &self.spec,
                     )
-                    .map_err(|e| {
-                        error!(
-                            self.log,
-                            "Error whilst signing attestation";
-                            "error" => format!("{:?}", e)
-                        )
-                    })
-                    .ok()?;
+                })?
+                .map_err(Error::UnableToSignAttestation)?;
 
                 metrics::inc_counter_vec(&metrics::SIGNED_ATTESTATIONS_TOTAL, &[metrics::SUCCESS]);
 
-                Some(())
+                Ok(())
             }
             Ok(Safe::SameData) => {
                 warn!(
@@ -301,7 +444,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                     &metrics::SIGNED_ATTESTATIONS_TOTAL,
                     &[metrics::SAME_DATA],
                 );
-                None
+                Err(Error::SameData)
             }
             Err(NotSafe::UnregisteredValidator(pk)) => {
                 warn!(
@@ -314,7 +457,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                     &metrics::SIGNED_ATTESTATIONS_TOTAL,
                     &[metrics::UNREGISTERED],
                 );
-                None
+                Err(Error::Slashable(NotSafe::UnregisteredValidator(pk)))
             }
             Err(e) => {
                 crit!(
@@ -327,7 +470,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
                     &metrics::SIGNED_ATTESTATIONS_TOTAL,
                     &[metrics::SLASHABLE],
                 );
-                None
+                Err(Error::Slashable(e))
             }
         }
     }
@@ -338,46 +481,64 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
     /// modified by actors other than the signing validator.
     pub fn produce_signed_aggregate_and_proof(
         &self,
-        validator_pubkey: &PublicKeyBytes,
+        validator_pubkey: PublicKeyBytes,
         validator_index: u64,
         aggregate: Attestation<E>,
         selection_proof: SelectionProof,
-    ) -> Option<SignedAggregateAndProof<E>> {
-        let validators = self.validators.read();
-        let voting_keypair = &validators.voting_keypair(validator_pubkey)?;
+    ) -> Result<SignedAggregateAndProof<E>, Error> {
+        // Take the fork early to avoid lock interleaving.
+        let fork = self.fork();
+
+        let proof = self.with_validator_keypair(validator_pubkey, move |keypair| {
+            SignedAggregateAndProof::from_aggregate(
+                validator_index,
+                aggregate,
+                Some(selection_proof),
+                &keypair.sk,
+                &fork,
+                self.genesis_validators_root,
+                &self.spec,
+            )
+        })?;
 
         metrics::inc_counter_vec(&metrics::SIGNED_AGGREGATES_TOTAL, &[metrics::SUCCESS]);
 
-        Some(SignedAggregateAndProof::from_aggregate(
-            validator_index,
-            aggregate,
-            Some(selection_proof),
-            &voting_keypair.sk,
-            &self.fork(),
-            self.genesis_validators_root,
-            &self.spec,
-        ))
+        Ok(proof)
     }
 
     /// Produces a `SelectionProof` for the `slot`, signed by with corresponding secret key to
     /// `validator_pubkey`.
     pub fn produce_selection_proof(
         &self,
-        validator_pubkey: &PublicKeyBytes,
+        validator_pubkey: PublicKeyBytes,
         slot: Slot,
-    ) -> Option<SelectionProof> {
-        let validators = self.validators.read();
-        let voting_keypair = &validators.voting_keypair(validator_pubkey)?;
+    ) -> Result<SelectionProof, Error> {
+        // Take the fork early to avoid lock interleaving.
+        let fork = self.fork();
+
+        // Bypass the `with_validator_keypair` function.
+        //
+        // This is because we don't care about doppelganger protection when it comes to selection
+        // proofs. They are not slashable and we need them to subscribe to subnets on the BN.
+        //
+        // As long as we disallow `SignedAggregateAndProof` then these selection proofs will never
+        // be published on the network.
+        let validators_lock = self.validators.read();
+        let keypair = validators_lock
+            .voting_keypair(&validator_pubkey)
+            .ok_or(Error::UnknownPubkey(validator_pubkey))?;
+
+        let proof = SelectionProof::new::<E>(
+            slot,
+            &keypair.sk,
+            &fork,
+            self.genesis_validators_root,
+            &self.spec,
+        );
 
         metrics::inc_counter_vec(&metrics::SIGNED_SELECTION_PROOFS_TOTAL, &[metrics::SUCCESS]);
 
-        Some(SelectionProof::new::<E>(
-            slot,
-            &voting_keypair.sk,
-            &self.fork(),
-            self.genesis_validators_root,
-            &self.spec,
-        ))
+        Ok(proof)
     }
 
     /// Prune the slashing protection database so that it remains performant.
@@ -411,10 +572,11 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
         let new_min_target_epoch = current_epoch.saturating_sub(SLASHING_PROTECTION_HISTORY_EPOCHS);
         let new_min_slot = new_min_target_epoch.start_slot(E::slots_per_epoch());
 
-        let validators = self.validators.read();
+        let all_pubkeys: Vec<_> = self.voting_pubkeys(DoppelgangerStatus::ignored);
+
         if let Err(e) = self
             .slashing_protection
-            .prune_all_signed_attestations(validators.iter_voting_pubkeys(), new_min_target_epoch)
+            .prune_all_signed_attestations(all_pubkeys.iter(), new_min_target_epoch)
         {
             error!(
                 self.log,
@@ -426,7 +588,7 @@ impl<T: SlotClock + 'static, E: EthSpec> ValidatorStore<T, E> {
 
         if let Err(e) = self
             .slashing_protection
-            .prune_all_signed_blocks(validators.iter_voting_pubkeys(), new_min_slot)
+            .prune_all_signed_blocks(all_pubkeys.iter(), new_min_slot)
         {
             error!(
                 self.log,