From 7e4fff683092fe3cbc3adbb77cef505409692819 Mon Sep 17 00:00:00 2001 From: viraladmin <00purple@gmail.com> Date: Fri, 3 Jul 2026 21:50:56 -0600 Subject: [PATCH] bootstrap fixes --- Cargo.lock | 4 +- Cargo.toml | 2 +- src/orphans/deep_sync_rollback.rs | 9 +- src/orphans/orphan_checkup.rs | 13 --- src/records/memory/connections.rs | 7 -- src/records/memory/network_mapping/monitor.rs | 91 +++++++++++++++++++ src/rpc/client/genesis_compat.rs | 46 ++++++++++ src/rpc/client/handshake_processing.rs | 22 +++++ src/rpc/client/mod.rs | 1 + src/rpc/command_maps.rs | 2 + src/rpc/commands/mod.rs | 2 + src/rpc/commands/network_monitor_add.rs | 9 ++ src/rpc/commands/network_monitor_remove.rs | 9 ++ src/rpc/commands/receive_torrent.rs | 23 ++++- src/rpc/server/handshake.rs | 38 +++++++- src/rpc/server/rpc_command_loop.rs | 30 ++++++ src/startup/network_broadcast.rs | 35 +++++++ 17 files changed, 315 insertions(+), 28 deletions(-) create mode 100644 src/rpc/client/genesis_compat.rs diff --git a/Cargo.lock b/Cargo.lock index c96e8d3..9a0f738 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -661,9 +661,9 @@ checksum = "a246d82be1c9d791c5dfde9a2bd045fc3cbba3fa2b11ad558f27d01712f00569" [[package]] name = "encrypted_images" -version = "1.4.0" +version = "1.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "761fe0fc7059c1e9cb4217f49263b146d269322354aeefb8fc519a60b2e8cef1" +checksum = "dda784ca648e472fc1bb0e408cb60b22e6b1647511d3c8f7144dadcf7133f605" dependencies = [ "base64 0.21.7", "encoding", diff --git a/Cargo.toml b/Cargo.toml index b196f12..7bf8a11 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -26,7 +26,7 @@ codegen-units = 1 [dependencies] rpassword = "7.1.0" -encrypted_images = "1.4.0" +encrypted_images = "1.4.1" skein = "0.1.0" chrono = "0.4" serde = { version = "1.0.108", features = ["derive"] } diff --git a/src/orphans/deep_sync_rollback.rs b/src/orphans/deep_sync_rollback.rs index f7385b5..2ca01d2 100644 --- a/src/orphans/deep_sync_rollback.rs +++ b/src/orphans/deep_sync_rollback.rs @@ -28,6 +28,13 @@ pub async fn deep_sync_rollback(mut params: CheckUp, wallet: Arc) { // A hash mismatch this far behind means the local // chain must step back until the torrents agree. if local_torrent.info.info_hash != remote_torrent.info.info_hash { + if params.local_height == 0 { + warn!( + "[orphan][deep_sync_rollback] remote genesis differs from local genesis; refusing to roll back below genesis" + ); + break; + } + if !params.node_syncing { begin_reorg_lock().await; } @@ -45,7 +52,7 @@ pub async fn deep_sync_rollback(mut params: CheckUp, wallet: Arc) { undo_transactions(undo_transactions_params, wallet.clone()) .await .ok(); - params.local_height -= 1; + params.local_height = params.local_height.saturating_sub(1); } else { // Matching torrents mark the shared ancestor. break; diff --git a/src/orphans/orphan_checkup.rs b/src/orphans/orphan_checkup.rs index d6becda..8bfaa1f 100644 --- a/src/orphans/orphan_checkup.rs +++ b/src/orphans/orphan_checkup.rs @@ -9,7 +9,6 @@ use crate::records::memory::torrent_status::{ }; use crate::records::unpack_block::load_by_binary_data::load_block_from_binary; use crate::records::unpack_block::unpack_header::load_block_header; -use crate::rpc::client::block_hash_vote::request_block_hash_at_height; use crate::torrent::structs::{DownloadSave, Torrent}; use crate::torrent::torrenting_system::create_file::combine_pieces; use crate::torrent::torrenting_system::download_locks::acquire_candidate_download; @@ -106,18 +105,6 @@ async fn candidate_attaches_before_rollback( // Metadata may choose a candidate, but only downloaded block bytes can // prove the rollback is safe. torrent.verify(height, ¶ms.db, wallet).await?; - let peer_canonical_hash = request_block_hash_at_height( - params.stream.clone(), - params.map.clone(), - params.connections_key.clone(), - height, - ) - .await?; - if peer_canonical_hash != torrent.info.block_hash { - return Err(format!( - "Staged candidate is not peer canonical at height {height}." - )); - } let _download_guard = acquire_candidate_download(height, &torrent.info.info_hash, true).await?; diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index 93537db..fee450c 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -823,13 +823,6 @@ fn spawn_monitor_update(ip: String, action: u8, monitored_address: String, port: } else { NodeInfo::remove_monitor(params).await }; - NodeInfo::broadcast_address_state( - context.map.clone(), - &monitored_address, - "", - &format!("{ip}:{port}"), - ) - .await; }); } diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index 9ead624..d3bbfdf 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -1,9 +1,15 @@ use super::*; +use crate::records::memory::response_channels::reserve_transient_entry_with_context; use crate::records::wallet_registry::resolve_pubkey_from_short_address; +use crate::rpc::command_maps::{RPC_NETWORK_MONITOR_ADD, RPC_NETWORK_MONITOR_REMOVE}; pub const MONITOR_ACTION_ADD: u8 = 1; pub const MONITOR_ACTION_REMOVE: u8 = 2; +lazy_static! { + static ref MONITOR_EVENTS_SEEN: Mutex> = Mutex::new(HashMap::new()); +} + impl NodeInfo { pub async fn monitor_signature( action: u8, @@ -99,12 +105,91 @@ impl NodeInfo { Self::apply_monitor(params, MONITOR_ACTION_REMOVE).await } + fn monitor_event_key(edit: &SignedMonitorEdit) -> String { + format!( + "{}:{}:{}:{}:{}:{}", + edit.action, + edit.monitored_address, + edit.monitoring_address, + edit.target_ip, + edit.modified_timestamp, + edit.modified_signature + ) + } + + async fn remember_monitor_event(edit: &SignedMonitorEdit) -> bool { + let key = Self::monitor_event_key(edit); + let mut seen = MONITOR_EVENTS_SEEN.lock().await; + let now = Utc::now().timestamp_millis() as u64; + seen.retain(|_, timestamp| now.saturating_sub(*timestamp) < 3_600_000); + seen.insert(key, now).is_none() + } + + async fn broadcast_monitor_event( + map: Arc>, + edit: &SignedMonitorEdit, + source_ip: &str, + ) { + let message_type = match edit.action { + MONITOR_ACTION_ADD => RPC_NETWORK_MONITOR_ADD, + MONITOR_ACTION_REMOVE => RPC_NETWORK_MONITOR_REMOVE, + _ => return, + }; + let monitored_bytes = match Wallet::short_address_to_bytes(&edit.monitored_address) { + Some(bytes) => bytes, + None => return, + }; + let monitoring_bytes = match Wallet::short_address_to_bytes(&edit.monitoring_address) { + Some(bytes) => bytes, + None => return, + }; + let target_ip_bytes = ip_to_binary(&edit.target_ip); + let timestamp_bytes = edit.modified_timestamp.to_le_bytes(); + let signature_bytes = match decode(&edit.modified_signature) { + Ok(bytes) => bytes, + Err(_) => return, + }; + let streams = { + let connections_lock = CONNECTIONS.read().await; + connections_lock + .as_ref() + .map(|connection| connection.get_all_ready_peer_streams_with_keys()) + .unwrap_or_default() + }; + + for (peer_key, unlocked_stream) in streams { + let Some((peer_ip, _)) = peer_key.rsplit_once(':') else { + continue; + }; + if !source_ip.is_empty() && peer_ip == source_ip { + continue; + } + let hashmap_key = reserve_transient_entry_with_context( + map.clone(), + Some(message_type), + Some(peer_key.clone()), + ) + .await; + let mut message: Vec = Vec::new(); + message.push(message_type); + message.extend_from_slice(&hashmap_key); + message.push(edit.action); + message.extend_from_slice(&monitored_bytes); + message.extend_from_slice(&monitoring_bytes); + message.extend_from_slice(&target_ip_bytes); + message.extend_from_slice(×tamp_bytes); + message.extend_from_slice(&signature_bytes); + RpcResponse::send_raw(&unlocked_stream, Some(&peer_key), &message).await; + } + } + async fn apply_monitor(params: MonitorAddressParams, action: u8) -> RpcResponse { let MonitorAddressParams { mut edit, remote_ip, db, wallet, + map, .. } = params; let current_timestamp = Utc::now().timestamp_millis() as u64; @@ -131,6 +216,11 @@ impl NodeInfo { return RpcResponse::Binary(b"Error: Could not validate monitor signature".to_vec()); } + let is_new_event = Self::remember_monitor_event(&edit).await; + if !is_new_event { + return RpcResponse::Binary(b"Success".to_vec()); + } + { let mut address_map = ADDRESS_MAP.lock().await; let Some(monitored) = address_map.get_mut(&edit.monitored_address) else { @@ -169,6 +259,7 @@ impl NodeInfo { } Self::persist_recovery_snapshot("monitor update").await; + Self::broadcast_monitor_event(map, &edit, &remote_ip).await; RpcResponse::Binary(b"Success".to_vec()) } diff --git a/src/rpc/client/genesis_compat.rs b/src/rpc/client/genesis_compat.rs new file mode 100644 index 0000000..be95682 --- /dev/null +++ b/src/rpc/client/genesis_compat.rs @@ -0,0 +1,46 @@ +use crate::common::check_genesis::genesis_checkup; +use crate::records::memory::response_channels::Command; +use crate::records::unpack_block::unpack_header::load_block_header; +use crate::rpc::client::block_hash_vote::request_block_hash_at_height; +use crate::{Arc, Mutex, TcpStream}; + +async fn local_genesis_hash() -> Result, String> { + if !genesis_checkup().await { + return Ok(None); + } + + let header = load_block_header(0).await?; + Ok(Some(header.hash().await)) +} + +pub async fn ensure_compatible_genesis( + stream: Arc>, + map: Arc>, + connections_key: String, + remote_height: u32, +) -> Result<(), String> { + let local_hash = local_genesis_hash().await?; + let remote_hash = + match request_block_hash_at_height(stream, map, connections_key.clone(), 0).await { + Ok(hash) => Some(hash), + Err(err) if err.contains("Block 0 not found") => None, + Err(err) => { + return Err(format!( + "remote genesis compatibility check failed for {connections_key}: {err}" + )); + } + }; + + match (local_hash, remote_hash) { + (Some(local), Some(remote)) if local != remote => Err(format!( + "incompatible genesis from peer {connections_key}: local={local} remote={remote}" + )), + (Some(_), None) => Err(format!( + "peer {connections_key} has no genesis while local chain already has one" + )), + (None, None) if remote_height > 0 => Err(format!( + "peer {connections_key} reports height {remote_height} but has no genesis" + )), + _ => Ok(()), + } +} diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index 213b1c4..7e6e1bd 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -21,6 +21,7 @@ use crate::records::memory::connections::{ use crate::records::memory::enums::{ClientType, ConnectionType}; use crate::records::memory::response_channels::{reserve_entry, Command}; use crate::records::memory::structs::{Connection, StoreConnectionParams}; +use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::client::handshake::connect_and_handshake; use crate::rpc::client::register_wallet::register_connected_wallet; use crate::rpc::client::structs::{Connect, Handshake}; @@ -170,6 +171,13 @@ pub async fn bootstrap_peer_discovery(mut params: BootstrapParams) -> Result<(), let local_height = get_height(¶ms.db); let remote_height = request_remote_height(stream.clone(), params.map.clone(), current_key.clone()).await?; + ensure_compatible_genesis( + stream.clone(), + params.map.clone(), + current_key.clone(), + remote_height, + ) + .await?; let local_genesis_exists = genesis_checkup().await; if !local_genesis_exists || remote_height > local_height + 10 { @@ -390,6 +398,20 @@ pub async fn process_handshake_response( "Remote height lookup failed before network join: {err}" )) })?; + ensure_compatible_genesis( + broadcast_stream.clone(), + params.map.clone(), + connections_key.clone(), + remote_height, + ) + .await + .map_err(|err| { + let connections_key = connections_key.clone(); + tokio::spawn(async move { + remove_key_from_memory(&connections_key).await; + }); + io::Error::other(format!("Genesis compatibility check failed: {err}")) + })?; let self_add_block = crate::records::memory::network_mapping::SELF_ADD_BLOCK; let self_add_limit = crate::records::memory::network_mapping::NodeInfo::self_add_limit_height(); let join_mode = if crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height( diff --git a/src/rpc/client/mod.rs b/src/rpc/client/mod.rs index 93c2f86..dfe6876 100644 --- a/src/rpc/client/mod.rs +++ b/src/rpc/client/mod.rs @@ -1,5 +1,6 @@ // The rpc client module contains the standalone client-side handshake and sync helpers. pub mod block_hash_vote; +pub mod genesis_compat; pub mod handshake; pub mod handshake_message; pub mod handshake_processing; diff --git a/src/rpc/command_maps.rs b/src/rpc/command_maps.rs index 70e0e58..062bd09 100644 --- a/src/rpc/command_maps.rs +++ b/src/rpc/command_maps.rs @@ -60,6 +60,8 @@ pub const RPC_LATEST_ADDRESS_TRANSACTIONS: u8 = 46; pub const RPC_MARKETING_CAMPAIGN_HISTORY: u8 = 47; pub const RPC_TOKEN_CATALOG: u8 = 48; pub const RPC_COLLATERAL_STATUS: u8 = 49; +pub const RPC_NETWORK_MONITOR_ADD: u8 = 50; +pub const RPC_NETWORK_MONITOR_REMOVE: u8 = 51; pub const RPC_REPLY: u8 = 255; pub const MAX_RPC_REPLY_BYTES: usize = 64 * 1024 * 1024; diff --git a/src/rpc/commands/mod.rs b/src/rpc/commands/mod.rs index d59e4d0..6eb2fe4 100644 --- a/src/rpc/commands/mod.rs +++ b/src/rpc/commands/mod.rs @@ -19,6 +19,8 @@ pub mod latest_block; pub mod marketing_campaign_history; pub mod memory_by_signature; pub mod network_info; +pub mod network_monitor_add; +pub mod network_monitor_remove; pub mod nft_list; pub mod nft_lookup; pub mod random_node; diff --git a/src/rpc/commands/network_monitor_add.rs b/src/rpc/commands/network_monitor_add.rs index 1e66dc0..52e30f5 100644 --- a/src/rpc/commands/network_monitor_add.rs +++ b/src/rpc/commands/network_monitor_add.rs @@ -1,3 +1,4 @@ +use crate::records::memory::enums::ClientType; use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit}; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::Command; @@ -15,6 +16,7 @@ pub async fn network_monitor_add( db: &Db, wallet: Arc, map: Arc>, + client_type: ClientType, ) -> Result<(u32, RpcResponse), String> { let (uid, _) = read_bytes_from_stream::read_uid_from_stream(connections_key, stream_locked.clone()) @@ -45,6 +47,13 @@ pub async fn network_monitor_add( .await?; let remote_ip = read_bytes_from_stream::read_caller_ip(stream_locked).await?; + if client_type != ClientType::Miner { + return Ok(( + uid, + RpcResponse::Binary(b"Error: monitor updates are miner-peer only".to_vec()), + )); + } + let result = NodeInfo::add_monitor(MonitorAddressParams { map, edit: SignedMonitorEdit { diff --git a/src/rpc/commands/network_monitor_remove.rs b/src/rpc/commands/network_monitor_remove.rs index d40339f..4420026 100644 --- a/src/rpc/commands/network_monitor_remove.rs +++ b/src/rpc/commands/network_monitor_remove.rs @@ -1,3 +1,4 @@ +use crate::records::memory::enums::ClientType; use crate::records::memory::network_mapping::structs::{MonitorAddressParams, SignedMonitorEdit}; use crate::records::memory::network_mapping::NodeInfo; use crate::records::memory::response_channels::Command; @@ -15,6 +16,7 @@ pub async fn network_monitor_remove( db: &Db, wallet: Arc, map: Arc>, + client_type: ClientType, ) -> Result<(u32, RpcResponse), String> { let (uid, _) = read_bytes_from_stream::read_uid_from_stream(connections_key, stream_locked.clone()) @@ -45,6 +47,13 @@ pub async fn network_monitor_remove( .await?; let remote_ip = read_bytes_from_stream::read_caller_ip(stream_locked).await?; + if client_type != ClientType::Miner { + return Ok(( + uid, + RpcResponse::Binary(b"Error: monitor updates are miner-peer only".to_vec()), + )); + } + let result = NodeInfo::remove_monitor(MonitorAddressParams { map, edit: SignedMonitorEdit { diff --git a/src/rpc/commands/receive_torrent.rs b/src/rpc/commands/receive_torrent.rs index 356a0ba..0a1da6f 100644 --- a/src/rpc/commands/receive_torrent.rs +++ b/src/rpc/commands/receive_torrent.rs @@ -305,9 +305,26 @@ pub async fn torrent_submission( } Err(e) => { if is_stale_next_height_error(&e) { - info!( - "[broadcast] ignored stale downloaded block: height={height} err={e}" - ); + let current_height = get_height(&db_clone); + if height > 0 && within_orphan_window(current_height, height) { + warn!( + "[broadcast] stale downloaded block is inside orphan window: height={height} local_height={current_height} err={e}" + ); + trigger_orphan_check( + &e, + height, + stream_clone, + &db_clone, + wallet_clone, + map_clone, + connections_key_clone, + ) + .await; + } else { + info!( + "[broadcast] ignored stale downloaded block: height={height} err={e}" + ); + } return; } error!("[broadcast] setup_download error: height={height} err={e}"); diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index dd54662..c5561cb 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -14,6 +14,7 @@ use crate::records::memory::connections::{ use crate::records::memory::response_channels::generate_uid; use crate::records::memory::response_channels::Command; use crate::records::memory::structs::Connection; +use crate::rpc::client::genesis_compat::ensure_compatible_genesis; use crate::rpc::client::register_wallet::register_connected_wallet; use crate::rpc::client::syncing::node_syncing; use crate::rpc::client::wallet_registry_sync::sync_wallet_registry_with_retries; @@ -25,7 +26,9 @@ use crate::rpc::server::structs::{CombineAndSendDataParams, HandshakeTestParams} use crate::rpc::server::tests::{endpoint_port, is_port_open}; use crate::sled::Db; use crate::sleep; -use crate::startup::network_broadcast::announce_self_to_network; +use crate::startup::network_broadcast::{ + announce_self_to_network, get_network_mapping_for_address, +}; use crate::startup::remote_height::request_remote_height; use crate::wallets::structures::Wallet; use crate::Arc; @@ -63,6 +66,13 @@ async fn sync_incoming_peer_before_operational( let local_height = get_height(db); let remote_height = request_remote_height(stream.clone(), map.clone(), connections_key.to_string()).await?; + ensure_compatible_genesis( + stream.clone(), + map.clone(), + connections_key.to_string(), + remote_height, + ) + .await?; let local_genesis_exists = genesis_checkup().await; if local_genesis_exists && remote_height < local_height { @@ -222,6 +232,18 @@ async fn complete_incoming_miner_setup( return; } }; + if let Err(err) = ensure_compatible_genesis( + stream.clone(), + map.clone(), + connections_key.to_string(), + remote_height, + ) + .await + { + error!("[startup] incoming peer genesis compatibility check failed: {err}"); + remove_key_from_memory(connections_key).await; + return; + } let self_add_allowed = crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height( remote_height, @@ -247,6 +269,20 @@ async fn complete_incoming_miner_setup( remove_key_from_memory(connections_key).await; return; } + if let Err(err) = get_network_mapping_for_address( + stream.clone(), + map.clone(), + db, + wallet.clone(), + connections_key, + &short_address, + ) + .await + { + error!("[startup] incoming peer local network record import failed: {err}"); + remove_key_from_memory(connections_key).await; + return; + } } mark_peer_network_map_synced(connections_key).await; diff --git a/src/rpc/server/rpc_command_loop.rs b/src/rpc/server/rpc_command_loop.rs index be1bf86..93b90c5 100644 --- a/src/rpc/server/rpc_command_loop.rs +++ b/src/rpc/server/rpc_command_loop.rs @@ -567,6 +567,36 @@ pub async fn start_loop( .send(&stream_locked, Some(&connections_key), uid) .await; } + 50 => { + // signed monitor-add event from a miner peer + let (uid, result) = commands::network_monitor_add::network_monitor_add( + &connections_key, + stream_locked.clone(), + &db, + wallet.clone(), + map.clone(), + client_type, + ) + .await?; + result + .send(&stream_locked, Some(&connections_key), uid) + .await; + } + 51 => { + // signed monitor-remove event from a miner peer + let (uid, result) = commands::network_monitor_remove::network_monitor_remove( + &connections_key, + stream_locked.clone(), + &db, + wallet.clone(), + map.clone(), + client_type, + ) + .await?; + result + .send(&stream_locked, Some(&connections_key), uid) + .await; + } 30 => { // request node list let (uid, _) = read_bytes_from_stream::read_uid_from_stream( diff --git a/src/startup/network_broadcast.rs b/src/startup/network_broadcast.rs index 435e5b7..5c61681 100644 --- a/src/startup/network_broadcast.rs +++ b/src/startup/network_broadcast.rs @@ -105,6 +105,34 @@ pub async fn get_network_mapping( db: &Db, _wallet: Arc, connections_key: &str, +) -> Result<(), String> { + get_network_mapping_inner(unlocked_stream, command_map, db, connections_key, None).await +} + +pub async fn get_network_mapping_for_address( + unlocked_stream: Arc>, + command_map: Arc>, + db: &Db, + _wallet: Arc, + connections_key: &str, + only_address: &str, +) -> Result<(), String> { + get_network_mapping_inner( + unlocked_stream, + command_map, + db, + connections_key, + Some(only_address), + ) + .await +} + +async fn get_network_mapping_inner( + unlocked_stream: Arc>, + command_map: Arc>, + db: &Db, + connections_key: &str, + only_address: Option<&str>, ) -> Result<(), String> { // request the remote peer's serialized node list and // import each advertised add/delete record locally @@ -177,6 +205,13 @@ pub async fn get_network_mapping( } } + if only_address + .map(|wanted_address| wanted_address != address) + .unwrap_or(false) + { + continue; + } + // Import signed map records as remote state. This verifies the original // signer but does not require the newly connecting node to be eligible // to add nodes locally after the mature-network gate.