Validate the genesis config downloaded over RPC before accepting it (bp #8474) (#8481)

automerge

core/src/validator.rs

@@ -43,6 +43,7 @@ use solana_sdk::{
     timing::timestamp,
 };
 use std::{
+    collections::HashSet,
     net::{IpAddr, Ipv4Addr, SocketAddr},
     path::{Path, PathBuf},
     process,
@@ -72,6 +73,7 @@ pub struct ValidatorConfig {
     pub fixed_leader_schedule: Option<FixedSchedule>,
     pub wait_for_supermajority: bool,
     pub new_hard_forks: Option<Vec<Slot>>,
+    pub trusted_validators: Option<HashSet<Pubkey>>, // None = trust all
 }
 
 impl Default for ValidatorConfig {
@@ -94,6 +96,7 @@ impl Default for ValidatorConfig {
             fixed_leader_schedule: None,
             wait_for_supermajority: false,
             new_hard_forks: None,
+            trusted_validators: None,
         }
     }
 }
@@ -345,9 +348,7 @@ impl Validator {
         }
 
         if let Some(snapshot_hash) = snapshot_hash {
-            if let Some(ref trusted_validators) =
-                config.snapshot_config.as_ref().unwrap().trusted_validators
-            {
+            if let Some(ref trusted_validators) = config.trusted_validators {
                 let mut trusted = false;
                 for _ in 0..10 {
                     trusted = cluster_info

core/tests/bank_forks.rs

@@ -55,7 +55,6 @@ mod tests {
             snapshot_interval_slots,
             snapshot_package_output_path: PathBuf::from(snapshot_output_path.path()),
             snapshot_path: PathBuf::from(snapshot_dir.path()),
-            trusted_validators: None,
         };
         bank_forks.set_snapshot_config(Some(snapshot_config.clone()));
         SnapshotTestConfig {

ledger-tool/src/main.rs

@@ -536,7 +536,6 @@ fn load_bank_forks(
             snapshot_interval_slots: 0, // Value doesn't matter
             snapshot_package_output_path: ledger_path.clone(),
             snapshot_path: ledger_path.clone().join("snapshot"),
-            trusted_validators: None,
         })
     };
     let account_paths = if let Some(account_paths) = arg_matches.value_of("account_paths") {

ledger/src/bank_forks.rs

@@ -6,7 +6,7 @@ use log::*;
 use solana_measure::measure::Measure;
 use solana_metrics::inc_new_counter_info;
 use solana_runtime::{bank::Bank, status_cache::MAX_CACHE_ENTRIES};
-use solana_sdk::{clock::Slot, pubkey::Pubkey, timing};
+use solana_sdk::{clock::Slot, timing};
 use std::{
     collections::{HashMap, HashSet},
     ops::Index,
@@ -26,10 +26,6 @@ pub struct SnapshotConfig {
 
     // Where to place the snapshots for recent slots
     pub snapshot_path: PathBuf,
-
-    // Validators that must vouch for a given snapshot hash before it's accepted
-    // None = accept any snapshot hash
-    pub trusted_validators: Option<HashSet<Pubkey>>,
 }
 
 #[derive(Error, Debug)]

local-cluster/tests/local_cluster.rs

@@ -726,12 +726,9 @@ fn test_snapshots_blockstore_floor() {
     let (cluster_nodes, _) = discover_cluster(&cluster.entry_point_info.gossip, 1).unwrap();
     let mut trusted_validators = HashSet::new();
     trusted_validators.insert(cluster_nodes[0].id);
-    if let Some(ref mut config) = validator_snapshot_test_config
+    validator_snapshot_test_config
         .validator_config
-        .snapshot_config
-    {
-        config.trusted_validators = Some(trusted_validators);
-    }
+        .trusted_validators = Some(trusted_validators);
 
     cluster.add_validator(
         &validator_snapshot_test_config.validator_config,
@@ -1012,7 +1009,6 @@ fn setup_snapshot_validator_config(
         snapshot_interval_slots,
         snapshot_package_output_path: PathBuf::from(snapshot_output_path.path()),
         snapshot_path: PathBuf::from(snapshot_dir.path()),
-        trusted_validators: None,
     };
 
     // Create the account paths

validator/src/main.rs

@@ -27,6 +27,7 @@ use solana_ledger::bank_forks::SnapshotConfig;
 use solana_perf::recycler::enable_recycler_warming;
 use solana_sdk::{
     clock::Slot,
+    genesis_config::GenesisConfig,
     hash::Hash,
     pubkey::Pubkey,
     signature::{Keypair, Signer},
@@ -35,7 +36,7 @@ use std::{
     collections::HashSet,
     fs::{self, File},
     io::{self, Read},
-    net::{SocketAddr, TcpListener},
+    net::{SocketAddr, TcpListener, UdpSocket},
     path::{Path, PathBuf},
     process::exit,
     str::FromStr,
@@ -79,39 +80,25 @@ fn new_spinner_progress_bar() -> ProgressBar {
     progress_bar
 }
 
-fn download_tar_bz2(
-    rpc_addr: &SocketAddr,
-    archive_name: &str,
-    download_path: &Path,
-    is_snapshot: bool,
-) -> Result<(), String> {
-    let archive_path = download_path.join(archive_name);
-    if archive_path.is_file() {
-        return Ok(());
+fn download_file(url: &str, destination_file: &Path, not_found_ok: bool) -> Result<(), String> {
+    if destination_file.is_file() {
+        return Err(format!("{:?} already exists", destination_file));
     }
-    fs::create_dir_all(download_path).map_err(|err| err.to_string())?;
-
-    let temp_archive_path = {
-        let mut p = archive_path.clone();
-        p.set_extension(".tmp");
-        p
-    };
-
-    let url = format!("http://{}/{}", rpc_addr, archive_name);
     let download_start = Instant::now();
 
+    fs::create_dir_all(destination_file.parent().unwrap()).map_err(|err| err.to_string())?;
+
+    let temp_destination_file = destination_file.with_extension(".tmp");
+
     let progress_bar = new_spinner_progress_bar();
     progress_bar.set_message(&format!("{}Downloading {}...", TRUCK, url));
 
     let client = reqwest::blocking::Client::new();
-    let response = client
-        .get(url.as_str())
-        .send()
-        .map_err(|err| err.to_string())?;
+    let response = client.get(url).send().map_err(|err| err.to_string())?;
 
-    if is_snapshot && response.status() == reqwest::StatusCode::NOT_FOUND {
+    if response.status() == reqwest::StatusCode::NOT_FOUND && not_found_ok {
         progress_bar.finish_and_clear();
-        warn!("Snapshot not found at {}", url);
+        info!("Archive not found at {}", url);
         return Ok(());
     }
 
@@ -158,10 +145,10 @@ fn download_tar_bz2(
         response,
     };
 
-    let mut file = File::create(&temp_archive_path)
-        .map_err(|err| format!("Unable to create {:?}: {:?}", temp_archive_path, err))?;
+    let mut file = File::create(&temp_destination_file)
+        .map_err(|err| format!("Unable to create {:?}: {:?}", temp_destination_file, err))?;
     std::io::copy(&mut source, &mut file)
-        .map_err(|err| format!("Unable to write {:?}: {:?}", temp_archive_path, err))?;
+        .map_err(|err| format!("Unable to write {:?}: {:?}", temp_destination_file, err))?;
 
     source.progress_bar.finish_and_clear();
     info!(
@@ -175,28 +162,32 @@ fn download_tar_bz2(
         )
     );
 
-    if !is_snapshot {
-        info!("Extracting {:?}...", archive_path);
-        let extract_start = Instant::now();
-        let tar_bz2 = File::open(&temp_archive_path)
-            .map_err(|err| format!("Unable to open {}: {:?}", archive_name, err))?;
-        let tar = BzDecoder::new(std::io::BufReader::new(tar_bz2));
-        let mut archive = tar::Archive::new(tar);
-        archive
-            .unpack(download_path)
-            .map_err(|err| format!("Unable to unpack {}: {:?}", archive_name, err))?;
-        info!(
-            "Extracted {} in {:?}",
-            archive_name,
-            Instant::now().duration_since(extract_start)
-        );
-    }
-    std::fs::rename(temp_archive_path, archive_path)
+    std::fs::rename(temp_destination_file, destination_file)
         .map_err(|err| format!("Unable to rename: {:?}", err))?;
 
     Ok(())
 }
 
+fn extract_archive(archive_filename: &Path, destination_dir: &Path) -> Result<(), String> {
+    info!("Extracting {:?}...", archive_filename);
+    let extract_start = Instant::now();
+
+    fs::create_dir_all(destination_dir).map_err(|err| err.to_string())?;
+    let tar_bz2 = File::open(&archive_filename)
+        .map_err(|err| format!("Unable to open {:?}: {:?}", archive_filename, err))?;
+    let tar = BzDecoder::new(std::io::BufReader::new(tar_bz2));
+    let mut archive = tar::Archive::new(tar);
+    archive
+        .unpack(destination_dir)
+        .map_err(|err| format!("Unable to unpack {:?}: {:?}", archive_filename, err))?;
+    info!(
+        "Extracted {:?} in {:?}",
+        archive_filename,
+        Instant::now().duration_since(extract_start)
+    );
+    Ok(())
+}
+
 fn get_shred_rpc_peers(
     cluster_info: &Arc<RwLock<ClusterInfo>>,
     expected_shred_version: Option<u16>,
@@ -227,7 +218,7 @@ fn get_shred_rpc_peers(
 
 fn get_trusted_snapshot_hashes(
     cluster_info: &Arc<RwLock<ClusterInfo>>,
-    trusted_validators: Option<&HashSet<Pubkey>>,
+    trusted_validators: &Option<HashSet<Pubkey>>,
 ) -> Option<HashSet<(Slot, Hash)>> {
     if let Some(trusted_validators) = trusted_validators {
         let mut trusted_snapshot_hashes = HashSet::new();
@@ -248,14 +239,11 @@ fn get_trusted_snapshot_hashes(
     }
 }
 
-fn get_rpc_node(
-    node: &Node,
+fn start_gossip_spy(
     identity_keypair: &Arc<Keypair>,
     entrypoint_gossip: &SocketAddr,
-    expected_shred_version: Option<u16>,
-    trusted_validators: Option<&HashSet<Pubkey>>,
-    snapshot_not_required: bool,
-) -> (ContactInfo, RpcClient, Option<(Slot, Hash)>) {
+    gossip_socket: UdpSocket,
+) -> (Arc<RwLock<ClusterInfo>>, Arc<AtomicBool>, GossipService) {
     let mut cluster_info = ClusterInfo::new(
         ClusterInfo::spy_contact_info(&identity_keypair.pubkey()),
         identity_keypair.clone(),
@@ -267,30 +255,59 @@ fn get_rpc_node(
     let gossip_service = GossipService::new(
         &cluster_info.clone(),
         None,
-        node.sockets.gossip.try_clone().unwrap(),
+        gossip_socket,
         &gossip_exit_flag,
     );
+    (cluster_info, gossip_exit_flag, gossip_service)
+}
 
-    let (rpc_contact_info, rpc_client, selected_snapshot_hash) = loop {
+fn get_rpc_node(
+    cluster_info: &Arc<RwLock<ClusterInfo>>,
+    validator_config: &ValidatorConfig,
+    blacklisted_rpc_nodes: &mut HashSet<Pubkey>,
+    snapshot_not_required: bool,
+) -> (ContactInfo, Option<(Slot, Hash)>) {
+    let mut blacklist_timeout = Instant::now();
+    loop {
         info!(
             "Searching for an RPC service, shred version={:?}...",
-            expected_shred_version
+            validator_config.expected_shred_version
         );
         sleep(Duration::from_secs(1));
         info!("\n{}", cluster_info.read().unwrap().contact_info_trace());
 
-        let rpc_peers = get_shred_rpc_peers(&cluster_info, expected_shred_version);
-        if rpc_peers.is_empty() {
-            info!("No RPC services found");
+        let rpc_peers = get_shred_rpc_peers(&cluster_info, validator_config.expected_shred_version);
+        let rpc_peers_total = rpc_peers.len();
+
+        // Filter out blacklisted nodes
+        let rpc_peers: Vec<_> = rpc_peers
+            .into_iter()
+            .filter(|rpc_peer| !blacklisted_rpc_nodes.contains(&rpc_peer.id))
+            .collect();
+        let rpc_peers_blacklisted = rpc_peers_total - rpc_peers.len();
+
+        info!(
+            "Total {} RPC nodes found. {} blacklisted ",
+            rpc_peers_total, rpc_peers_blacklisted
+        );
+
+        if rpc_peers_blacklisted == rpc_peers_total {
+            // If all nodes are blacklisted and no additional nodes are discovered after 60 seconds,
+            // remove the blacklist and try them all again
+            if blacklist_timeout.elapsed().as_secs() > 60 {
+                info!("Blacklist timeout expired");
+                blacklisted_rpc_nodes.clear();
+            }
             continue;
         }
+        blacklist_timeout = Instant::now();
 
         let mut highest_snapshot_hash: Option<(Slot, Hash)> = None;
         let eligible_rpc_peers = if snapshot_not_required {
             rpc_peers
         } else {
             let trusted_snapshot_hashes =
-                get_trusted_snapshot_hashes(&cluster_info, trusted_validators);
+                get_trusted_snapshot_hashes(&cluster_info, &validator_config.trusted_validators);
 
             let mut eligible_rpc_peers = vec![];
 
@@ -330,8 +347,14 @@ fn get_rpc_node(
                 }
                 Some(highest_snapshot_hash) => {
                     info!(
-                        "Highest available snapshot slot is {}, available from {:?}",
+                        "Highest available snapshot slot is {}, available from {} node{}: {:?}",
                         highest_snapshot_hash.0,
+                        eligible_rpc_peers.len(),
+                        if eligible_rpc_peers.len() > 1 {
+                            "s"
+                        } else {
+                            ""
+                        },
                         eligible_rpc_peers
                             .iter()
                             .map(|contact_info| contact_info.id)
@@ -343,37 +366,11 @@ fn get_rpc_node(
         };
 
         if !eligible_rpc_peers.is_empty() {
-            // Prefer the entrypoint's RPC service if present, otherwise pick one at random
-            let contact_info = if let Some(contact_info) = eligible_rpc_peers
-                .iter()
-                .find(|contact_info| contact_info.gossip == *entrypoint_gossip)
-            {
-                contact_info
-            } else {
-                &eligible_rpc_peers[thread_rng().gen_range(0, eligible_rpc_peers.len())]
-            };
-
-            info!(
-                "Trying RPC service from node {}: {:?}",
-                contact_info.id, contact_info.rpc
-            );
-            let rpc_client = RpcClient::new_socket(contact_info.rpc);
-            match rpc_client.get_version() {
-                Ok(rpc_version) => {
-                    info!("RPC node version: {}", rpc_version.solana_core);
-                    break (contact_info.clone(), rpc_client, highest_snapshot_hash);
-                }
-                Err(err) => {
-                    warn!("Failed to get RPC node's version: {}", err);
-                }
-            }
+            let contact_info =
+                &eligible_rpc_peers[thread_rng().gen_range(0, eligible_rpc_peers.len())];
+            return (contact_info.clone(), highest_snapshot_hash);
         }
-    };
-
-    gossip_exit_flag.store(true, Ordering::Relaxed);
-    gossip_service.join().unwrap();
-
-    (rpc_contact_info, rpc_client, selected_snapshot_hash)
+    }
 }
 
 fn check_vote_account(
@@ -426,29 +423,78 @@ fn check_vote_account(
     Ok(())
 }
 
-fn download_ledger(
+fn download_genesis(
+    rpc_addr: &SocketAddr,
+    ledger_path: &Path,
+    validator_config: &mut ValidatorConfig,
+) -> Result<(), String> {
+    let genesis_package = ledger_path.join("genesis.tar.bz2");
+
+    let genesis_config = if !genesis_package.exists() {
+        let tmp_genesis_path = ledger_path.join("tmp-genesis");
+        let tmp_genesis_package = tmp_genesis_path.join("genesis.tar.bz2");
+
+        let _ignored = fs::remove_dir_all(&tmp_genesis_path);
+        download_file(
+            &format!("http://{}/{}", rpc_addr, "genesis.tar.bz2"),
+            &tmp_genesis_package,
+            false,
+        )?;
+        extract_archive(&tmp_genesis_package, &ledger_path)?;
+
+        let tmp_genesis_config = GenesisConfig::load(&ledger_path)
+            .map_err(|err| format!("Failed to load downloaded genesis config: {}", err))?;
+
+        if let Some(expected_genesis_hash) = validator_config.expected_genesis_hash {
+            if expected_genesis_hash != tmp_genesis_config.hash() {
+                return Err(format!(
+                    "Genesis hash mismatch: expected {} but downloaded genesis hash is {}",
+                    expected_genesis_hash,
+                    tmp_genesis_config.hash(),
+                ));
+            }
+        }
+
+        std::fs::rename(tmp_genesis_package, genesis_package)
+            .map_err(|err| format!("Unable to rename: {:?}", err))?;
+        tmp_genesis_config
+    } else {
+        GenesisConfig::load(&ledger_path)
+            .map_err(|err| format!("Failed to load genesis config: {}", err))?
+    };
+
+    if validator_config.expected_genesis_hash.is_none() {
+        info!("Expected genesis hash set to {}", genesis_config.hash());
+        // If no particular genesis hash is expected use the one that's here
+        validator_config.expected_genesis_hash = Some(genesis_config.hash());
+    }
+    Ok(())
+}
+
+fn download_snapshot(
     rpc_addr: &SocketAddr,
     ledger_path: &Path,
     snapshot_hash: Option<(Slot, Hash)>,
 ) -> Result<(), String> {
-    download_tar_bz2(rpc_addr, "genesis.tar.bz2", ledger_path, false)?;
-
-    if snapshot_hash.is_some() {
-        let snapshot_package =
-            solana_ledger::snapshot_utils::get_snapshot_archive_path(ledger_path);
-        if snapshot_package.exists() {
-            fs::remove_file(&snapshot_package)
-                .map_err(|err| format!("error removing {:?}: {}", snapshot_package, err))?;
-        }
-        download_tar_bz2(
-            rpc_addr,
-            snapshot_package.file_name().unwrap().to_str().unwrap(),
-            snapshot_package.parent().unwrap(),
-            true,
-        )
-        .map_err(|err| format!("Failed to fetch snapshot: {:?}", err))?;
+    if snapshot_hash.is_none() {
+        return Ok(());
     }
 
+    let snapshot_package = solana_ledger::snapshot_utils::get_snapshot_archive_path(ledger_path);
+    if snapshot_package.exists() {
+        fs::remove_file(&snapshot_package)
+            .map_err(|err| format!("error removing {:?}: {}", snapshot_package, err))?;
+    }
+    download_file(
+        &format!(
+            "http://{}/{}",
+            rpc_addr,
+            snapshot_package.file_name().unwrap().to_str().unwrap()
+        ),
+        &snapshot_package,
+        true,
+    )?;
+
     Ok(())
 }
 
@@ -821,6 +867,7 @@ pub fn main() {
             .map(|rpc_port| (rpc_port, rpc_port + 1)),
         voting_disabled: matches.is_present("no_voting"),
         wait_for_supermajority: !matches.is_present("no_wait_for_supermajority"),
+        trusted_validators,
         ..ValidatorConfig::default()
     };
 
@@ -869,7 +916,6 @@ pub fn main() {
         },
         snapshot_path,
         snapshot_package_output_path: ledger_path.clone(),
-        trusted_validators,
     });
 
     if matches.is_present("limit_ledger_size") {
@@ -1027,55 +1073,88 @@ pub fn main() {
         );
 
         if !no_genesis_fetch {
-            let (rpc_contact_info, rpc_client, snapshot_hash) = get_rpc_node(
-                &node,
+            let (cluster_info, gossip_exit_flag, gossip_service) = start_gossip_spy(
                 &identity_keypair,
                 &cluster_entrypoint.gossip,
-                validator_config.expected_shred_version,
-                validator_config
-                    .snapshot_config
-                    .as_ref()
-                    .unwrap()
-                    .trusted_validators
-                    .as_ref(),
-                no_snapshot_fetch,
+                node.sockets.gossip.try_clone().unwrap(),
             );
 
-            let genesis_hash = rpc_client.get_genesis_hash().unwrap_or_else(|err| {
-                error!("Failed to get genesis hash: {}", err);
-                exit(1);
-            });
+            let mut blacklisted_rpc_nodes = HashSet::new();
+            loop {
+                let (rpc_contact_info, snapshot_hash) = get_rpc_node(
+                    &cluster_info,
+                    &validator_config,
+                    &mut blacklisted_rpc_nodes,
+                    no_snapshot_fetch,
+                );
+                info!(
+                    "Using RPC service from node {}: {:?}",
+                    rpc_contact_info.id, rpc_contact_info.rpc
+                );
+                let rpc_client = RpcClient::new_socket(rpc_contact_info.rpc);
 
-            if let Some(expected_genesis_hash) = validator_config.expected_genesis_hash {
-                if expected_genesis_hash != genesis_hash {
-                    error!(
-                        "Genesis hash mismatch: expected {} but local genesis hash is {}",
-                        expected_genesis_hash, genesis_hash,
-                    );
-                    exit(1);
+                let result = match rpc_client.get_version() {
+                    Ok(rpc_version) => {
+                        info!("RPC node version: {}", rpc_version.solana_core);
+                        Ok(())
+                    }
+                    Err(err) => Err(format!("Failed to get RPC node version: {}", err)),
                 }
-            }
-            validator_config.expected_genesis_hash = Some(genesis_hash);
+                .and_then(|_| {
+                    download_genesis(
+                        &rpc_contact_info.rpc,
+                        &ledger_path,
+                        &mut validator_config,
+                    )
+                })
+                .and_then(|_| {
+                    if let Some(expected_genesis_hash) = validator_config.expected_genesis_hash {
+                        // Sanity check that the RPC node is using the expected genesis hash before
+                        // downloading a snapshot from it
+                        let rpc_genesis_hash = rpc_client
+                            .get_genesis_hash()
+                            .map_err(|err| format!("Failed to get genesis hash: {}", err))?;
 
-            if !validator_config.voting_disabled && !no_check_vote_account {
-                check_vote_account(
-                    &rpc_client,
-                    &vote_account,
-                    &voting_keypair.pubkey(),
-                    &identity_keypair.pubkey(),
-                )
-                .unwrap_or_else(|err| {
-                    error!("Failed to check vote account: {}", err);
-                    exit(1);
+                        if expected_genesis_hash != rpc_genesis_hash {
+                            return Err(format!("Genesis hash mismatch: expected {} but RPC node genesis hash is {}",
+                                               expected_genesis_hash, rpc_genesis_hash));
+                        }
+                    }
+                    Ok(())
+                })
+                .and_then(|_| download_snapshot(&rpc_contact_info.rpc, &ledger_path, snapshot_hash))
+                .and_then(|_| {
+                    if !validator_config.voting_disabled && !no_check_vote_account {
+                        check_vote_account(
+                            &rpc_client,
+                            &vote_account,
+                            &voting_keypair.pubkey(),
+                            &identity_keypair.pubkey(),
+                        )
+                    } else {
+                        Ok(())
+                    }
                 });
-            }
 
-            download_ledger(&rpc_contact_info.rpc, &ledger_path, snapshot_hash).unwrap_or_else(
-                |err| {
-                    error!("Failed to initialize ledger: {}", err);
-                    exit(1);
-                },
-            );
+                if result.is_ok() {
+                    break;
+                }
+                warn!("{}", result.unwrap_err());
+
+                if let Some(ref trusted_validators) = validator_config.trusted_validators {
+                    if trusted_validators.contains(&rpc_contact_info.id) {
+                        continue; // Never blacklist a trusted node
+                    }
+                }
+
+                info!(
+                    "Excluding {} as a future RPC candidate",
+                    rpc_contact_info.id
+                );
+                blacklisted_rpc_nodes.insert(rpc_contact_info.id);
+            }
+            gossip_exit_flag.store(true, Ordering::Relaxed);
+            gossip_service.join().unwrap();
         }
     }