Weak subjectivity start from genesis (#1675)

## Issue Addressed
Solution 2 proposed here: https://github.com/sigp/lighthouse/issues/1435#issuecomment-692317639

## Proposed Changes
- Adds an optional `--wss-checkpoint` flag that takes a string `root:epoch`
- Verify that the given checkpoint exists in the chain, or that the the chain syncs through this checkpoint. If not, shutdown and prompt the user to purge state before restarting.

## Additional Info


Co-authored-by: Paul Hauner <paul@paulhauner.com>

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -12,7 +12,6 @@ use crate::errors::{BeaconChainError as Error, BlockProductionError};
 use crate::eth1_chain::{Eth1Chain, Eth1ChainBackend};
 use crate::events::{EventHandler, EventKind};
 use crate::head_tracker::HeadTracker;
-use crate::metrics;
 use crate::migrate::Migrate;
 use crate::naive_aggregation_pool::{Error as NaiveAggregationError, NaiveAggregationPool};
 use crate::observed_attestations::{Error as AttestationObservationError, ObservedAttestations};
@@ -27,7 +26,9 @@ use crate::timeout_rw_lock::TimeoutRwLock;
 use crate::validator_pubkey_cache::ValidatorPubkeyCache;
 use crate::BeaconForkChoiceStore;
 use crate::BeaconSnapshot;
+use crate::{metrics, BeaconChainError};
 use fork_choice::ForkChoice;
+use futures::channel::mpsc::Sender;
 use itertools::process_results;
 use operation_pool::{OperationPool, PersistedOperationPool};
 use parking_lot::RwLock;
@@ -222,6 +223,9 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub(crate) validator_pubkey_cache: TimeoutRwLock<ValidatorPubkeyCache>,
     /// A list of any hard-coded forks that have been disabled.
     pub disabled_forks: Vec<String>,
+    /// Sender given to tasks, so that if they encounter a state in which execution cannot
+    /// continue they can request that everything shuts down.
+    pub shutdown_sender: Sender<&'static str>,
     /// Logging to CLI, etc.
     pub(crate) log: Logger,
     /// Arbitrary bytes included in the blocks.
@@ -680,7 +684,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
     /// Returns the block canonical root of the current canonical chain at a given slot.
     ///
-    /// Returns None if a block doesn't exist at the slot.
+    /// Returns `None` if the given slot doesn't exist in the chain.
     pub fn root_at_slot(&self, target_slot: Slot) -> Result<Option<Hash256>, Error> {
         process_results(self.rev_iter_block_roots()?, |mut iter| {
             iter.find(|(_, slot)| *slot == target_slot)
@@ -688,6 +692,26 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         })
     }
 
+    /// Returns the block canonical root of the current canonical chain at a given slot, starting from the given state.
+    ///
+    /// Returns `None` if the given slot doesn't exist in the chain.
+    pub fn root_at_slot_from_state(
+        &self,
+        target_slot: Slot,
+        beacon_block_root: Hash256,
+        state: &BeaconState<T::EthSpec>,
+    ) -> Result<Option<Hash256>, Error> {
+        let iter = BlockRootsIterator::new(self.store.clone(), state);
+        let iter_with_head = std::iter::once(Ok((beacon_block_root, state.slot)))
+            .chain(iter)
+            .map(|result| result.map_err(|e| e.into()));
+
+        process_results(iter_with_head, |mut iter| {
+            iter.find(|(_, slot)| *slot == target_slot)
+                .map(|(root, _)| root)
+        })
+    }
+
     /// Returns the block proposer for a given slot.
     ///
     /// Information is read from the present `beacon_state` shuffling, only information from the
@@ -1242,7 +1266,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     return ChainSegmentResult::Failed {
                         imported_blocks,
                         error: BlockError::NotFinalizedDescendant { block_parent_root },
-                    }
+                    };
                 }
                 // If there was an error whilst determining if the block was invalid, return that
                 // error.
@@ -1250,7 +1274,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     return ChainSegmentResult::Failed {
                         imported_blocks,
                         error: BlockError::BeaconChainError(e),
-                    }
+                    };
                 }
                 // If the block was decided to be irrelevant for any other reason, don't include
                 // this block or any of it's children in the filtered chain segment.
@@ -1284,7 +1308,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     return ChainSegmentResult::Failed {
                         imported_blocks,
                         error,
-                    }
+                    };
                 }
             };
 
@@ -1296,7 +1320,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         return ChainSegmentResult::Failed {
                             imported_blocks,
                             error,
-                        }
+                        };
                     }
                 }
             }
@@ -1514,6 +1538,38 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             check_block_is_finalized_descendant::<T, _>(signed_block, &fork_choice, &self.store)?;
         let block = &signed_block.message;
 
+        // compare the existing finalized checkpoint with the incoming block's finalized checkpoint
+        let old_finalized_checkpoint = fork_choice.finalized_checkpoint();
+        let new_finalized_checkpoint = state.finalized_checkpoint;
+
+        // Only perform the weak subjectivity check if it was configured.
+        if let Some(wss_checkpoint) = self.config.weak_subjectivity_checkpoint {
+            // This ensures we only perform the check once.
+            if (old_finalized_checkpoint.epoch < wss_checkpoint.epoch)
+                && (wss_checkpoint.epoch <= new_finalized_checkpoint.epoch)
+            {
+                if let Err(e) =
+                    self.verify_weak_subjectivity_checkpoint(wss_checkpoint, block_root, &state)
+                {
+                    let mut shutdown_sender = self.shutdown_sender();
+                    crit!(
+                        self.log,
+                        "Weak subjectivity checkpoint verification failed while importing block!";
+                        "block_root" => format!("{:?}", block_root),
+                        "parent_root" => format!("{:?}", block.parent_root),
+                        "old_finalized_epoch" => format!("{:?}", old_finalized_checkpoint.epoch),
+                        "new_finalized_epoch" => format!("{:?}", new_finalized_checkpoint.epoch),
+                        "weak_subjectivity_epoch" => format!("{:?}", wss_checkpoint.epoch),
+                        "error" => format!("{:?}", e),
+                    );
+                    crit!(self.log, "You must use the `--purge-db` flag to clear the database and restart sync. You may be on a hostile network.");
+                    shutdown_sender.try_send("Weak subjectivity checkpoint verification failed. Provided block root is not a checkpoint.")
+                        .map_err(|err|BlockError::BeaconChainError(BeaconChainError::WeakSubjectivtyShutdownError(err)))?;
+                    return Err(BlockError::WeakSubjectivityConflict);
+                }
+            }
+        }
+
         // Register the new block with the fork choice service.
         {
             let _fork_choice_block_timer =
@@ -1928,6 +1984,60 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         Ok(())
     }
 
+    /// This function takes a configured weak subjectivity `Checkpoint` and the latest finalized `Checkpoint`.
+    /// If the weak subjectivity checkpoint and finalized checkpoint share the same epoch, we compare
+    /// roots. If we the weak subjectivity checkpoint is from an older epoch, we iterate back through
+    /// roots in the canonical chain until we reach the finalized checkpoint from the correct epoch, and
+    /// compare roots. This must called on startup and during verification of any block which causes a finality
+    /// change affecting the weak subjectivity checkpoint.
+    pub fn verify_weak_subjectivity_checkpoint(
+        &self,
+        wss_checkpoint: Checkpoint,
+        beacon_block_root: Hash256,
+        state: &BeaconState<T::EthSpec>,
+    ) -> Result<(), BeaconChainError> {
+        let finalized_checkpoint = state.finalized_checkpoint;
+        info!(self.log, "Verifying the configured weak subjectivity checkpoint"; "weak_subjectivity_epoch" => wss_checkpoint.epoch, "weak_subjectivity_root" => format!("{:?}", wss_checkpoint.root));
+        // If epochs match, simply compare roots.
+        if wss_checkpoint.epoch == finalized_checkpoint.epoch
+            && wss_checkpoint.root != finalized_checkpoint.root
+        {
+            crit!(
+                self.log,
+                 "Root found at the specified checkpoint differs";
+                  "weak_subjectivity_root" => format!("{:?}", wss_checkpoint.root),
+                  "finalized_checkpoint_root" => format!("{:?}", finalized_checkpoint.root)
+            );
+            return Err(BeaconChainError::WeakSubjectivtyVerificationFailure);
+        } else if wss_checkpoint.epoch < finalized_checkpoint.epoch {
+            let slot = wss_checkpoint
+                .epoch
+                .start_slot(T::EthSpec::slots_per_epoch());
+
+            // Iterate backwards through block roots from the given state. If first slot of the epoch is a skip-slot,
+            // this will return the root of the closest prior non-skipped slot.
+            match self.root_at_slot_from_state(slot, beacon_block_root, state)? {
+                Some(root) => {
+                    if root != wss_checkpoint.root {
+                        crit!(
+                            self.log,
+                             "Root found at the specified checkpoint differs";
+                              "weak_subjectivity_root" => format!("{:?}", wss_checkpoint.root),
+                              "finalized_checkpoint_root" => format!("{:?}", finalized_checkpoint.root)
+                        );
+                        return Err(BeaconChainError::WeakSubjectivtyVerificationFailure);
+                    }
+                }
+                None => {
+                    crit!(self.log, "The root at the start slot of the given epoch could not be found";
+                    "wss_checkpoint_slot" => format!("{:?}", slot));
+                    return Err(BeaconChainError::WeakSubjectivtyVerificationFailure);
+                }
+            }
+        }
+        Ok(())
+    }
+
     /// Called by the timer on every slot.
     ///
     /// Performs slot-based pruning.
@@ -2163,6 +2273,11 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         writeln!(output, "}}").unwrap();
     }
 
+    /// Get a channel to request shutting down.
+    pub fn shutdown_sender(&self) -> Sender<&'static str> {
+        self.shutdown_sender.clone()
+    }
+
     // Used for debugging
     #[allow(dead_code)]
     pub fn dump_dot_file(&self, file_name: &str) {

beacon_node/beacon_chain/src/block_verification.rs

@@ -207,6 +207,13 @@ pub enum BlockError<T: EthSpec> {
     /// We were unable to process this block due to an internal error. It's unclear if the block is
     /// valid.
     BeaconChainError(BeaconChainError),
+    /// There was an error whilst verifying weak subjectivity. This block conflicts with the
+    /// configured weak subjectivity checkpoint and was not imported.
+    ///
+    /// ## Peer scoring
+    ///
+    /// The block is invalid and the peer is faulty.
+    WeakSubjectivityConflict,
 }
 
 impl<T: EthSpec> std::fmt::Display for BlockError<T> {

beacon_node/beacon_chain/src/builder.rs

@@ -18,6 +18,7 @@ use crate::{
 };
 use eth1::Config as Eth1Config;
 use fork_choice::ForkChoice;
+use futures::channel::mpsc::Sender;
 use operation_pool::{OperationPool, PersistedOperationPool};
 use parking_lot::RwLock;
 use slog::{info, Logger};
@@ -107,6 +108,7 @@ pub struct BeaconChainBuilder<T: BeaconChainTypes> {
     eth1_chain: Option<Eth1Chain<T::Eth1Chain, T::EthSpec>>,
     event_handler: Option<T::EventHandler>,
     slot_clock: Option<T::SlotClock>,
+    shutdown_sender: Option<Sender<&'static str>>,
     head_tracker: Option<HeadTracker>,
     data_dir: Option<PathBuf>,
     pubkey_cache_path: Option<PathBuf>,
@@ -154,6 +156,7 @@ where
             eth1_chain: None,
             event_handler: None,
             slot_clock: None,
+            shutdown_sender: None,
             head_tracker: None,
             pubkey_cache_path: None,
             data_dir: None,
@@ -405,6 +408,12 @@ where
         self
     }
 
+    /// Sets a `Sender` to allow the beacon chain to send shutdown signals.
+    pub fn shutdown_sender(mut self, sender: Sender<&'static str>) -> Self {
+        self.shutdown_sender = Some(sender);
+        self
+    }
+
     /// Creates a new, empty operation pool.
     fn empty_op_pool(mut self) -> Self {
         self.op_pool = Some(OperationPool::new());
@@ -575,6 +584,9 @@ where
             shuffling_cache: TimeoutRwLock::new(ShufflingCache::new()),
             validator_pubkey_cache: TimeoutRwLock::new(validator_pubkey_cache),
             disabled_forks: self.disabled_forks,
+            shutdown_sender: self
+                .shutdown_sender
+                .ok_or_else(|| "Cannot build without a shutdown sender.".to_string())?,
             log: log.clone(),
             graffiti: self.graffiti,
         };
@@ -583,6 +595,27 @@ where
             .head()
             .map_err(|e| format!("Failed to get head: {:?}", e))?;
 
+        // Only perform the check if it was configured.
+        if let Some(wss_checkpoint) = beacon_chain.config.weak_subjectivity_checkpoint {
+            if let Err(e) = beacon_chain.verify_weak_subjectivity_checkpoint(
+                wss_checkpoint,
+                head.beacon_block_root,
+                &head.beacon_state,
+            ) {
+                crit!(
+                    log,
+                    "Weak subjectivity checkpoint verification failed on startup!";
+                    "head_block_root" => format!("{}", head.beacon_block_root),
+                    "head_slot" => format!("{}", head.beacon_block.slot()),
+                    "finalized_epoch" => format!("{}", head.beacon_state.finalized_checkpoint.epoch),
+                    "wss_checkpoint_epoch" => format!("{}", wss_checkpoint.epoch),
+                    "error" => format!("{:?}", e),
+                );
+                crit!(log, "You must use the `--purge-db` flag to clear the database and restart sync. You may be on a hostile network.");
+                return Err(format!("Weak subjectivity verification failed: {:?}", e));
+            }
+        }
+
         info!(
             log,
             "Beacon chain initialized";
@@ -761,6 +794,8 @@ mod test {
         )
         .expect("should create interop genesis state");
 
+        let (shutdown_tx, _) = futures::channel::mpsc::channel(1);
+
         let chain = BeaconChainBuilder::new(MinimalEthSpec)
             .logger(log.clone())
             .store(Arc::new(store))
@@ -773,6 +808,7 @@ mod test {
             .null_event_handler()
             .testing_slot_clock(Duration::from_secs(1))
             .expect("should configure testing slot clock")
+            .shutdown_sender(shutdown_tx)
             .build()
             .expect("should build");
 

beacon_node/beacon_chain/src/chain_config.rs

@@ -1,4 +1,5 @@
 use serde_derive::{Deserialize, Serialize};
+use types::Checkpoint;
 
 /// There is a 693 block skip in the current canonical Medalla chain, we use 700 to be safe.
 pub const DEFAULT_IMPORT_BLOCK_MAX_SKIP_SLOTS: u64 = 700;
@@ -10,12 +11,17 @@ pub struct ChainConfig {
     ///
     /// If `None`, there is no limit.
     pub import_max_skip_slots: Option<u64>,
+    /// A user-input `Checkpoint` that must exist in the beacon chain's sync path.
+    ///
+    /// If `None`, there is no weak subjectivity verification.
+    pub weak_subjectivity_checkpoint: Option<Checkpoint>,
 }
 
 impl Default for ChainConfig {
     fn default() -> Self {
         Self {
             import_max_skip_slots: Some(DEFAULT_IMPORT_BLOCK_MAX_SKIP_SLOTS),
+            weak_subjectivity_checkpoint: None,
         }
     }
 }

beacon_node/beacon_chain/src/errors.rs

@@ -5,6 +5,7 @@ use crate::naive_aggregation_pool::Error as NaiveAggregationError;
 use crate::observed_attestations::Error as ObservedAttestationsError;
 use crate::observed_attesters::Error as ObservedAttestersError;
 use crate::observed_block_producers::Error as ObservedBlockProducersError;
+use futures::channel::mpsc::TrySendError;
 use operation_pool::OpPoolError;
 use safe_arith::ArithError;
 use ssz_types::Error as SszTypesError;
@@ -83,6 +84,8 @@ pub enum BeaconChainError {
     ObservedBlockProducersError(ObservedBlockProducersError),
     PruningError(PruningError),
     ArithError(ArithError),
+    WeakSubjectivtyVerificationFailure,
+    WeakSubjectivtyShutdownError(TrySendError<&'static str>),
 }
 
 easy_from_to!(SlotProcessingError, BeaconChainError);

beacon_node/beacon_chain/src/test_utils.rs

@@ -8,8 +8,9 @@ use crate::{
     builder::{BeaconChainBuilder, Witness},
     eth1_chain::CachingEth1Backend,
     events::NullEventHandler,
-    BeaconChain, BeaconChainTypes, StateSkipConfig,
+    BeaconChain, BeaconChainTypes, BlockError, ChainConfig, StateSkipConfig,
 };
+use futures::channel::mpsc::Receiver;
 use genesis::interop_genesis_state;
 use rand::rngs::StdRng;
 use rand::Rng;
@@ -107,6 +108,7 @@ pub struct BeaconChainHarness<T: BeaconChainTypes> {
     pub chain: BeaconChain<T>,
     pub spec: ChainSpec,
     pub data_dir: TempDir,
+    pub shutdown_receiver: Receiver<&'static str>,
 
     pub rng: StdRng,
 }
@@ -134,6 +136,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorEphemeralHarnessType<E>> {
 
         let config = StoreConfig::default();
         let store = Arc::new(HotColdDB::open_ephemeral(config, spec.clone(), log.clone()).unwrap());
+        let (shutdown_tx, shutdown_receiver) = futures::channel::mpsc::channel(1);
 
         let chain = BeaconChainBuilder::new(eth_spec_instance)
             .logger(log.clone())
@@ -151,6 +154,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorEphemeralHarnessType<E>> {
             .null_event_handler()
             .testing_slot_clock(HARNESS_SLOT_TIME)
             .unwrap()
+            .shutdown_sender(shutdown_tx)
             .build()
             .unwrap();
 
@@ -159,6 +163,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorEphemeralHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -184,7 +189,25 @@ impl<E: EthSpec> BeaconChainHarness<NullMigratorEphemeralHarnessType<E>> {
         eth_spec_instance: E,
         validators_keypairs: Vec<Keypair>,
         target_aggregators_per_committee: u64,
-        config: StoreConfig,
+        store_config: StoreConfig,
+    ) -> Self {
+        Self::new_with_chain_config(
+            eth_spec_instance,
+            validators_keypairs,
+            target_aggregators_per_committee,
+            store_config,
+            ChainConfig::default(),
+        )
+    }
+
+    /// Instantiate a new harness with `validator_count` initial validators, a custom
+    /// `target_aggregators_per_committee` spec value, and a `ChainConfig`
+    pub fn new_with_chain_config(
+        eth_spec_instance: E,
+        validators_keypairs: Vec<Keypair>,
+        target_aggregators_per_committee: u64,
+        store_config: StoreConfig,
+        chain_config: ChainConfig,
     ) -> Self {
         let data_dir = tempdir().expect("should create temporary data_dir");
         let mut spec = E::default_spec();
@@ -195,8 +218,9 @@ impl<E: EthSpec> BeaconChainHarness<NullMigratorEphemeralHarnessType<E>> {
         let drain = slog_term::FullFormat::new(decorator).build();
         let debug_level = slog::LevelFilter::new(drain, slog::Level::Debug);
         let log = slog::Logger::root(std::sync::Mutex::new(debug_level).fuse(), o!());
+        let (shutdown_tx, shutdown_receiver) = futures::channel::mpsc::channel(1);
 
-        let store = HotColdDB::open_ephemeral(config, spec.clone(), log.clone()).unwrap();
+        let store = HotColdDB::open_ephemeral(store_config, spec.clone(), log.clone()).unwrap();
         let chain = BeaconChainBuilder::new(eth_spec_instance)
             .logger(log)
             .custom_spec(spec.clone())
@@ -213,6 +237,8 @@ impl<E: EthSpec> BeaconChainHarness<NullMigratorEphemeralHarnessType<E>> {
             .null_event_handler()
             .testing_slot_clock(HARNESS_SLOT_TIME)
             .expect("should configure testing slot clock")
+            .shutdown_sender(shutdown_tx)
+            .chain_config(chain_config)
             .build()
             .expect("should build");
 
@@ -221,6 +247,7 @@ impl<E: EthSpec> BeaconChainHarness<NullMigratorEphemeralHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -240,6 +267,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
         let drain = slog_term::FullFormat::new(decorator).build();
         let debug_level = slog::LevelFilter::new(drain, slog::Level::Debug);
         let log = slog::Logger::root(std::sync::Mutex::new(debug_level).fuse(), o!());
+        let (shutdown_tx, shutdown_receiver) = futures::channel::mpsc::channel(1);
 
         let chain = BeaconChainBuilder::new(eth_spec_instance)
             .logger(log.clone())
@@ -258,6 +286,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
             .null_event_handler()
             .testing_slot_clock(HARNESS_SLOT_TIME)
             .expect("should configure testing slot clock")
+            .shutdown_sender(shutdown_tx)
             .build()
             .expect("should build");
 
@@ -266,6 +295,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -282,6 +312,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
         let spec = E::default_spec();
 
         let log = NullLoggerBuilder.build().expect("logger should build");
+        let (shutdown_tx, shutdown_receiver) = futures::channel::mpsc::channel(1);
 
         let chain = BeaconChainBuilder::new(eth_spec_instance)
             .logger(log.clone())
@@ -300,6 +331,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
             .null_event_handler()
             .testing_slot_clock(Duration::from_secs(1))
             .expect("should configure testing slot clock")
+            .shutdown_sender(shutdown_tx)
             .build()
             .expect("should build");
 
@@ -308,6 +340,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -608,6 +641,17 @@ where
         block_hash
     }
 
+    pub fn process_block_result(
+        &self,
+        slot: Slot,
+        block: SignedBeaconBlock<E>,
+    ) -> Result<SignedBeaconBlockHash, BlockError<E>> {
+        assert_eq!(self.chain.slot().unwrap(), slot);
+        let block_hash: SignedBeaconBlockHash = self.chain.process_block(block)?.into();
+        self.chain.fork_choice().unwrap();
+        Ok(block_hash)
+    }
+
     pub fn process_attestations(&self, attestations: HarnessAttestations<E>) {
         for (unaggregated_attestations, maybe_signed_aggregate) in attestations.into_iter() {
             for (attestation, subnet_id) in unaggregated_attestations {