Weak subjectivity start from genesis (#1675)

This commit was edited by Paul H when rebasing from master to
v0.3.0-staging.

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

- 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.

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;
@@ -224,6 +225,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.
@@ -727,7 +731,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)
@@ -735,6 +739,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
@@ -1274,7 +1298,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.
@@ -1282,7 +1306,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.
@@ -1316,7 +1340,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                     return ChainSegmentResult::Failed {
                         imported_blocks,
                         error,
-                    }
+                    };
                 }
             };
 
@@ -1328,7 +1352,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
                         return ChainSegmentResult::Failed {
                             imported_blocks,
                             error,
-                        }
+                        };
                     }
                 }
             }
@@ -1537,6 +1561,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 =
@@ -1951,6 +2007,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.
@@ -2309,6 +2419,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,
@@ -410,6 +413,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());
@@ -581,6 +590,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,
         };
@@ -589,6 +601,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";
@@ -767,6 +800,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))
@@ -779,6 +814,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;
@@ -87,6 +88,8 @@ pub enum BeaconChainError {
         shuffling_epoch: Epoch,
         head_block_epoch: Epoch,
     },
+    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;
@@ -109,6 +110,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,
 }
@@ -136,6 +138,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())
@@ -153,6 +156,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorEphemeralHarnessType<E>> {
             .null_event_handler()
             .testing_slot_clock(HARNESS_SLOT_TIME)
             .unwrap()
+            .shutdown_sender(shutdown_tx)
             .build()
             .unwrap();
 
@@ -161,6 +165,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorEphemeralHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -186,7 +191,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();
@@ -197,8 +220,9 @@ impl<E: EthSpec> BeaconChainHarness<NullMigratorEphemeralHarnessType<E>> {
         let drain = slog_term::FullFormat::new(decorator).build();
         let debug_level = slog::LevelFilter::new(drain, slog::Level::Critical);
         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())
@@ -215,6 +239,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");
 
@@ -223,6 +249,7 @@ impl<E: EthSpec> BeaconChainHarness<NullMigratorEphemeralHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -242,6 +269,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
         let drain = slog_term::FullFormat::new(decorator).build();
         let debug_level = slog::LevelFilter::new(drain, slog::Level::Critical);
         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())
@@ -260,6 +288,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");
 
@@ -268,6 +297,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -284,6 +314,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())
@@ -302,6 +333,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");
 
@@ -310,6 +342,7 @@ impl<E: EthSpec> BeaconChainHarness<BlockingMigratorDiskHarnessType<E>> {
             chain,
             validators_keypairs,
             data_dir,
+            shutdown_receiver,
             rng: make_rng(),
         }
     }
@@ -697,6 +730,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 {