channelmanager: Create FundRecoverer to take our node in offline mode so that we can just send a BogusChannelReestablish and close all the StubChannelMonitors and sweep the funds from the events.

Author: Aditya Sharma

fuzz/src/chanmon_consistency.rs

@@ -303,6 +303,10 @@ impl chain::Watch<TestChannelSigner> for TestChainMonitor {
 	) -> Result<(), ()> {
 		return self.chain_monitor.panic_and_persist_stub_channel(funding_outpoint, stub_monitor);
 	}
+
+	fn get_stub_cids_with_counterparty(&self, counterparty_node_id: PublicKey) -> Vec<ChannelId> {
+		return self.chain_monitor.get_stub_cids_with_counterparty(counterparty_node_id);
+	}
 }
 
 struct KeyProvider {

lightning/src/chain/chainmonitor.rs

@@ -860,6 +860,17 @@ where C::Target: chain::Filter,
 		Ok(())
 	}
 
+	fn get_stub_cids_with_counterparty(&self, counterparty_node_id: PublicKey) -> Vec<ChannelId> {
+		let stub_monitors = self.stub_monitors.read().unwrap();
+		let mut stubs = vec![];
+		for (_, mon) in stub_monitors.iter() {
+			if mon.get_counterparty_node_id() == Some(counterparty_node_id) {
+				stubs.push(mon.channel_id());
+			}
+		}
+		stubs
+	}
+
 	fn stale_or_missing_channel_monitor(&self, stub_chan: &StubChannel) -> bool {
 		let monitors = self.monitors.read().unwrap();
 
@@ -1061,7 +1072,7 @@ impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref,
 	      L::Target: Logger,
 	      P::Target: Persist<ChannelSigner>,
 {
-	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
+	/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] & [`StubChannelMonitor`] upon maturity.
 	///
 	/// For channels featuring anchor outputs, this method will also process [`BumpTransaction`]
 	/// events produced from each [`ChannelMonitor`] while there is a balance to claim onchain
@@ -1083,6 +1094,14 @@ impl<ChannelSigner: EcdsaChannelSigner, C: Deref, T: Deref, F: Deref, L: Deref,
 				}
 			}
 		}
+		for stub_monitor_state in self.stub_monitors.read().unwrap().values() {
+			match stub_monitor_state.process_pending_events(&handler) {
+				Ok(()) => {},
+				Err(ReplayEvent ()) => {
+					self.event_notifier.notify();
+				}
+			}
+		}
 	}
 }
 

lightning/src/chain/mod.rs

@@ -345,6 +345,15 @@ pub trait Watch<ChannelSigner: EcdsaChannelSigner> {
 	/// - In test builds, instead of panicking, it delegates the handling of a missing channel to the
 	///   `watch_dummy` function.
 	fn panic_and_persist_stub_channel(&self, funding_outpoint: OutPoint, stub_monitor: StubChannelMonitor<ChannelSigner>) -> Result<(), ()>;
+
+	/// Retrieves a list of channel IDs for [`StubChannelMonitor`] associated with a specific counterparty node ID.
+	///
+	/// This function searches through the collection of [`StubChannelMonitor`] and collects the channel IDs
+	/// of those monitors that have the specified counterparty node ID.
+	///
+	/// This is used by [`FundRecoverer`] to fetch all the [`ChannelId`] with a peer that needs recovery so that we can send them
+	/// `BogusChannelReestablish`.
+	fn get_stub_cids_with_counterparty(&self, counterparty_node_id: PublicKey) -> Vec<ChannelId>;
 }
 
 /// The `Filter` trait defines behavior for indicating chain activity of interest pertaining to

lightning/src/ln/channelmanager.rs

@@ -2313,6 +2313,188 @@ where
 	logger: L,
 }
 
+
+/// This works as a mock [`ChannelMessageHandler`] it is used mainly when a user wants to run their node in
+/// offline mode i.e. This node won't communicate with any peer except sending a BogusChannelReestablish
+/// for all the [`StubChannelMonitors`] being tracked by the [`ChainMonitor`].
+/// 
+/// [`FundRecoverer`] is parameterized by a number of components to achieve this.
+/// - [`chain::Watch`] (typically [`ChainMonitor`]) for on-chain monitoring and enforcement of each
+///   channel
+/// - [`SignerProvider`] for providing signers whose operations are scoped to individual channels
+/// - [`Logger`] for logging operational information of varying degrees
+///
+/// Additionally, it implements the following traits:
+/// - [`ChannelMessageHandler`] to handle off-chain channel activity from peers
+/// - [`MessageSendEventsProvider`] to similarly send such messages to peers
+///
+pub struct FundRecoverer<SP: Deref, L:Deref, M: Deref> 
+	where
+	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
+	SP::Target: SignerProvider,
+	L::Target: Logger
+{	
+	default_configuration: UserConfig,
+	chain_monitor: M,
+	chain_hash: ChainHash,
+	per_peer_state: FairRwLock<HashMap<PublicKey, Mutex<PeerState<SP>>>>,
+	logger: L,
+}
+
+impl<SP:Deref, L:Deref, M:Deref> MessageSendEventsProvider for FundRecoverer<SP, L, M>
+	where
+	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
+	SP::Target: SignerProvider,
+	L::Target: Logger
+{
+	fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {
+		let mut pending_events = Vec::new();
+		let events = RefCell::new(Vec::new());
+		let per_peer_state = self.per_peer_state.read().unwrap();
+		for (_cp_id, peer_state_mutex) in per_peer_state.iter() {
+			let mut peer_state_lock = peer_state_mutex.lock().unwrap();
+			let peer_state = &mut *peer_state_lock;
+			if peer_state.pending_msg_events.len() > 0 {
+				pending_events.append(&mut peer_state.pending_msg_events);
+			}
+		}
+		if !pending_events.is_empty() {
+			events.replace(pending_events);
+		}
+		events.into_inner()
+	}
+}
+
+impl<SP:Deref, L: Deref, M:Deref> FundRecoverer<SP, L, M>
+	where
+	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
+	SP::Target: SignerProvider,
+	L::Target: Logger
+{
+	/// Creates a new instance of `FundRecoverer`.
+    ///
+    /// This function initializes a `FundRecoverer` with the provided `chain_monitor`,
+    /// `logger`, configuration, and chain parameters. The `FundRecoverer` is set up with
+    /// the default configuration and a chain hash derived from the genesis block of the
+    /// specified network.
+	pub fn new(chain_monitor: M, logger: L, config: UserConfig, params: ChainParameters) -> Self {
+		return Self { default_configuration: config.clone(),
+			chain_monitor, 
+			chain_hash: ChainHash::using_genesis_block(params.network),
+			per_peer_state: FairRwLock::new(new_hash_map()),
+			logger
+		}
+	}
+}
+
+impl<SP:Deref, L:Deref, M:Deref> ChannelMessageHandler for FundRecoverer<SP, L, M>
+	where
+	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
+	SP::Target: SignerProvider,
+	L::Target: Logger
+{
+	fn handle_open_channel(&self, _their_node_id: &PublicKey, _msg: &msgs::OpenChannel) {}
+	fn handle_accept_channel(&self, _their_node_id: &PublicKey, _msg: &msgs::AcceptChannel) {}
+	fn handle_funding_created(&self, _their_node_id: &PublicKey, _msg: &msgs::FundingCreated) {}
+	fn handle_funding_signed(&self, _their_node_id: &PublicKey, _msg: &msgs::FundingSigned) {}
+	fn handle_channel_ready(&self, _their_node_id: &PublicKey, _msg: &msgs::ChannelReady) {}
+	fn handle_shutdown(&self, _their_node_id: &PublicKey, _msg: &msgs::Shutdown) {}
+	fn handle_closing_signed(&self, _their_node_id: &PublicKey, _msg: &msgs::ClosingSigned) {}
+	fn handle_update_add_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateAddHTLC) {}
+	fn handle_update_fulfill_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFulfillHTLC) {}
+	fn handle_update_fail_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFailHTLC) {}
+	fn handle_update_fail_malformed_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFailMalformedHTLC) {}
+	fn handle_commitment_signed(&self, _their_node_id: &PublicKey, _msg: &msgs::CommitmentSigned) {}
+	fn handle_revoke_and_ack(&self, _their_node_id: &PublicKey, _msg: &msgs::RevokeAndACK) {}
+	fn handle_update_fee(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFee) {}
+	fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &msgs::AnnouncementSignatures) {}
+	fn handle_channel_update(&self, _their_node_id: &PublicKey, _msg: &msgs::ChannelUpdate) {}
+	fn handle_open_channel_v2(&self, _their_node_id: &PublicKey, _msg: &msgs::OpenChannelV2) {}
+	fn handle_accept_channel_v2(&self, _their_node_id: &PublicKey, _msg: &msgs::AcceptChannelV2) {}
+	fn handle_stfu(&self, _their_node_id: &PublicKey, _msg: &msgs::Stfu) {}
+	#[cfg(splicing)]
+	fn handle_splice_init(&self, _their_node_id: &PublicKey, _msg: &msgs::SpliceInit) {}
+	#[cfg(splicing)]
+	fn handle_splice_ack(&self, _their_node_id: &PublicKey, _msg: &msgs::SpliceAck) {}
+	#[cfg(splicing)]
+	fn handle_splice_locked(&self, _their_node_id: &PublicKey, _msg: &msgs::SpliceLocked) {}
+	fn handle_tx_add_input(&self, _their_node_id: &PublicKey, _msg: &msgs::TxAddInput) {}
+	fn handle_tx_add_output(&self, _their_node_id: &PublicKey, _msg: &msgs::TxAddOutput) {}
+	fn handle_tx_remove_input(&self, _their_node_id: &PublicKey, _msg: &msgs::TxRemoveInput) {}
+	fn handle_tx_remove_output(&self, _their_node_id: &PublicKey, _msg: &msgs::TxRemoveOutput) {}
+	fn handle_tx_complete(&self, _their_node_id: &PublicKey, _msg: &msgs::TxComplete) {}
+	fn handle_tx_signatures(&self, _their_node_id: &PublicKey, _msg: &msgs::TxSignatures) {}
+	fn handle_tx_init_rbf(&self, _their_node_id: &PublicKey, _msg: &msgs::TxInitRbf) {}
+	fn handle_tx_ack_rbf(&self, _their_node_id: &PublicKey, _msg: &msgs::TxAckRbf) {}
+	fn handle_tx_abort(&self, _their_node_id: &PublicKey, _msg: &msgs::TxAbort) {}
+	fn handle_peer_storage(&self, _their_node_id: &PublicKey, _msg: &msgs::PeerStorageMessage) {}
+	fn handle_your_peer_storage(&self, _their_node_id: &PublicKey, _msg: &msgs::YourPeerStorageMessage) {}
+	fn peer_disconnected(&self, _their_node_id: &PublicKey) {}
+
+	fn peer_connected(&self, counterparty_node_id: &PublicKey, init_msg: &msgs::Init, _inbound: bool) -> Result<(), ()> {
+		let logger = WithContext::from(&self.logger, Some(*counterparty_node_id), None, None);
+
+		{
+			let mut peer_state_lock = self.per_peer_state.write().unwrap();
+			match peer_state_lock.entry(counterparty_node_id.clone()) {
+				hash_map::Entry::Vacant(e) => {
+					e.insert(Mutex::new(PeerState {
+						channel_by_id: new_hash_map(),
+						inbound_channel_request_by_id: new_hash_map(),
+						latest_features: init_msg.features.clone(),
+						pending_msg_events: Vec::new(),
+						in_flight_monitor_updates: BTreeMap::new(),
+						monitor_update_blocked_actions: BTreeMap::new(),
+						actions_blocking_raa_monitor_updates: BTreeMap::new(),
+						is_connected: true,
+					}));
+				},
+				hash_map::Entry::Occupied(e) => {
+					let mut peer_state = e.get().lock().unwrap();
+					peer_state.latest_features = init_msg.features.clone();
+
+					debug_assert!(!peer_state.is_connected, "A peer shouldn't be connected twice");
+					peer_state.is_connected = true;
+				},
+			}
+		}
+		log_debug!(logger, "Generating Bogus channel_reestablish events for all the stub channels with peer {}", log_pubkey!(counterparty_node_id));
+
+		let per_peer_state = self.per_peer_state.read().unwrap();
+		if let Some(peer_state_mutex) = per_peer_state.get(counterparty_node_id) {
+			let mut peer_state_lock = peer_state_mutex.lock().unwrap();
+			let peer_state = &mut *peer_state_lock;
+			let pending_msg_events = &mut peer_state.pending_msg_events;
+			for cid in self.chain_monitor.get_stub_cids_with_counterparty(*counterparty_node_id) {
+				pending_msg_events.push(MessageSendEvent::SendChannelReestablish {
+					node_id: *counterparty_node_id,
+					msg: msgs::ChannelReestablish {
+						channel_id: cid,
+						next_local_commitment_number: 0,
+						next_remote_commitment_number: 0,
+						your_last_per_commitment_secret: [1u8; 32],
+						my_current_per_commitment_point: PublicKey::from_slice(&[2u8; 33]).unwrap(),
+						next_funding_txid: None,
+					},
+				})
+			}
+		}
+		Ok(())
+	}
+
+	fn handle_channel_reestablish(&self, _their_node_id: &PublicKey, _msg: &msgs::ChannelReestablish) {}
+	fn handle_error(&self, _their_node_id: &PublicKey, _msg: &msgs::ErrorMessage) {}
+	fn provided_node_features(&self) -> NodeFeatures {
+		provided_node_features(&self.default_configuration)
+	}
+	fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures {
+		provided_init_features(&self.default_configuration)
+	}
+	fn get_chain_hashes(&self) -> Option<Vec<ChainHash>> {
+		Some(vec![self.chain_hash])
+	}
+}
+
 /// Chain-related parameters used to construct a new `ChannelManager`.
 ///
 /// Typically, the block-specific parameters are derived from the best block hash for the network,

lightning/src/util/test_utils.rs

@@ -462,6 +462,9 @@ impl<'a> chain::Watch<TestChannelSigner> for TestChainMonitor<'a> {
 		return self.chain_monitor.panic_and_persist_stub_channel(funding_outpoint, stub_monitor);
 	}
 
+	fn get_stub_cids_with_counterparty(&self, counterparty_node_id: PublicKey) -> Vec<ChannelId> {
+		return self.chain_monitor.get_stub_cids_with_counterparty(counterparty_node_id);
+	}
 }
 
 #[cfg(test)]