Implement DisconnectPeer and HandleOutboundMsgGenerationError events

Add test for DisconnectPeer event
Update DisconnectPeer with optional ErrorMessage
Manage error for funding_transaction_generated
Add disconnect_socket to SocketDescriptor trait

Author: Antoine Riard

fuzz/fuzz_targets/full_stack_target.rs

@@ -107,12 +107,16 @@ impl BroadcasterInterface for TestBroadcaster {
 #[derive(Clone, PartialEq, Eq, Hash)]
 struct Peer {
 	id: u8,
+	connected: bool,
 }
 impl SocketDescriptor for Peer {
 	fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, _resume_read: bool) -> usize {
 		assert!(write_offset < data.len());
 		data.len() - write_offset
 	}
+	fn disconnect_socket(&mut self) {
+		self.connected = false;
+	}
 }
 
 #[inline]

src/ln/channel.rs

@@ -328,7 +328,7 @@ macro_rules! secp_call {
 		match $res {
 			Ok(key) => key,
 			//TODO: make the error a parameter
-			Err(_) => return Err(HandleError{err: $err, action: Some(msgs::ErrorAction::DisconnectPeer{})})
+			Err(_) => return Err(HandleError{err: $err, action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })})
 		}
 	};
 }
@@ -433,10 +433,10 @@ impl Channel {
 
 	fn check_remote_fee(fee_estimator: &FeeEstimator, feerate_per_kw: u32) -> Result<(), HandleError> {
 		if (feerate_per_kw as u64) < fee_estimator.get_est_sat_per_vbyte(ConfirmationTarget::Background) * 250 {
-			return Err(HandleError{err: "Peer's feerate much too low", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Peer's feerate much too low", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if (feerate_per_kw as u64) > fee_estimator.get_est_sat_per_vbyte(ConfirmationTarget::HighPriority) * 375 { // 375 = 250 * 1.5x
-			return Err(HandleError{err: "Peer's feerate much too high", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Peer's feerate much too high", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		Ok(())
 	}
@@ -448,29 +448,32 @@ impl Channel {
 	pub fn new_from_req(fee_estimator: &FeeEstimator, chan_keys: ChannelKeys, their_node_id: PublicKey, msg: &msgs::OpenChannel, user_id: u64, announce_publicly: bool) -> Result<Channel, HandleError> {
 		// Check sanity of message fields:
 		if msg.funding_satoshis >= MAX_FUNDING_SATOSHIS {
-			return Err(HandleError{err: "funding value > 2^24", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "funding value > 2^24", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if msg.channel_reserve_satoshis > msg.funding_satoshis {
-			return Err(HandleError{err: "Bogus channel_reserve_satoshis", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Bogus channel_reserve_satoshis", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if msg.push_msat > (msg.funding_satoshis - msg.channel_reserve_satoshis) * 1000 {
-			return Err(HandleError{err: "push_msat more than highest possible value", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "push_msat more than highest possible value", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if msg.dust_limit_satoshis > msg.funding_satoshis {
-			return Err(HandleError{err: "Peer never wants payout outputs?", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Peer never wants payout outputs?", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
+		}
+		if msg.max_htlc_value_in_flight_msat > msg.funding_satoshis * 1000 {
+			return Err(HandleError{err: "Bogus max_htlc_value_in_flight_satoshis", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if msg.htlc_minimum_msat >= (msg.funding_satoshis - msg.channel_reserve_satoshis) * 1000 {
-			return Err(HandleError{err: "Minimum htlc value is full channel value", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Minimum htlc value is full channel value", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		Channel::check_remote_fee(fee_estimator, msg.feerate_per_kw)?;
 		if msg.to_self_delay > MAX_LOCAL_BREAKDOWN_TIMEOUT {
-			return Err(HandleError{err: "They wanted our payments to be delayed by a needlessly long period", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "They wanted our payments to be delayed by a needlessly long period", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if msg.max_accepted_htlcs < 1 {
-			return Err(HandleError{err: "0 max_accpted_htlcs makes for a useless channel", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "0 max_accpted_htlcs makes for a useless channel", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		if (msg.channel_flags & 254) != 0 {
-			return Err(HandleError{err: "unknown channel_flags", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "unknown channel_flags", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 
 		// Convert things into internal flags and prep our state:

src/ln/channelmanager.rs

@@ -676,6 +676,16 @@ impl ChannelManager {
 	/// Call this upon creation of a funding transaction for the given channel.
 	/// Panics if a funding transaction has already been provided for this channel.
 	pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_txo: OutPoint) {
+
+		macro_rules! add_pending_event {
+			($event: expr) => {
+				{
+					let mut pending_events = self.pending_events.lock().unwrap();
+					pending_events.push($event);
+				}
+			}
+		}
+
 		let (chan, msg, chan_monitor) = {
 			let mut channel_state = self.channel_state.lock().unwrap();
 			match channel_state.by_id.remove(temporary_channel_id) {
@@ -684,10 +694,15 @@ impl ChannelManager {
 						Ok(funding_msg) => {
 							(chan, funding_msg.0, funding_msg.1)
 						},
-						Err(_e) => {
-							//TODO: Push e to pendingevents
+						Err(e) => {
+							mem::drop(channel_state);
+							add_pending_event!(events::Event::DisconnectPeer {
+								node_id: chan.get_their_node_id(),
+								msg: if let Some(msgs::ErrorAction::DisconnectPeer { msg } ) = e.action { msg } else { None },
+							});
+
 							return;
-						}
+						},
 					}
 				},
 				None => return
@@ -696,13 +711,10 @@ impl ChannelManager {
 		if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
 			unimplemented!(); // maybe remove from claimable_htlcs?
 		}
-		{
-			let mut pending_events = self.pending_events.lock().unwrap();
-			pending_events.push(events::Event::SendFundingCreated {
-				node_id: chan.get_their_node_id(),
-				msg: msg,
-			});
-		}
+		add_pending_event!(events::Event::SendFundingCreated {
+			node_id: chan.get_their_node_id(),
+			msg: msg,
+		});
 
 		let mut channel_state = self.channel_state.lock().unwrap();
 		channel_state.by_id.insert(chan.channel_id(), chan);

src/ln/msgs.rs

@@ -377,7 +377,9 @@ pub enum ErrorAction {
 		msg: UpdateFailHTLC
 	},
 	/// The peer took some action which made us think they were useless. Disconnect them.
-	DisconnectPeer,
+	DisconnectPeer {
+		msg: Option<ErrorMessage>
+	},
 	/// The peer did something harmless that we weren't able to process, just log and ignore
 	IgnoreError,
 	/// The peer did something incorrect. Tell them.

src/ln/peer_channel_encryptor.rs

@@ -147,7 +147,7 @@ impl PeerChannelEncryptor {
 
 		let mut chacha = ChaCha20Poly1305RFC::new(key, &nonce, h);
 		if !chacha.decrypt(&cyphertext[0..cyphertext.len() - 16], res, &cyphertext[cyphertext.len() - 16..]) {
-			return Err(HandleError{err: "Bad MAC", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Bad MAC", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 		Ok(())
 	}
@@ -195,11 +195,11 @@ impl PeerChannelEncryptor {
 		assert_eq!(act.len(), 50);
 
 		if act[0] != 0 {
-			return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+			return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 		}
 
 		let their_pub = match PublicKey::from_slice(secp_ctx, &act[1..34]) {
-			Err(_) => return Err(HandleError{err: "Invalid public key", action: Some(msgs::ErrorAction::DisconnectPeer{})}),
+			Err(_) => return Err(HandleError{err: "Invalid public key", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })}),
 			Ok(key) => key,
 		};
 
@@ -349,14 +349,14 @@ impl PeerChannelEncryptor {
 							panic!("Requested act at wrong step");
 						}
 						if act_three[0] != 0 {
-							return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+							return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
 						}
 
 						let mut their_node_id = [0; 33];
 						PeerChannelEncryptor::decrypt_with_ad(&mut their_node_id, 1, &temp_k2.unwrap(), &bidirectional_state.h, &act_three[1..50])?;
 						self.their_node_id = Some(match PublicKey::from_slice(&self.secp_ctx, &their_node_id) {
 							Ok(key) => key,
-							Err(_) => return Err(HandleError{err: "Bad node_id from peer", action: Some(msgs::ErrorAction::DisconnectPeer{})}),
+							Err(_) => return Err(HandleError{err: "Bad node_id from peer", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })}),
 						});
 
 						let mut sha = Sha256::new();

src/ln/peer_handler.rs

@@ -37,6 +37,12 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone {
 	/// indicating that read events on this descriptor should resume. A resume_read of false does
 	/// *not* imply that further read events should be paused.
 	fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, resume_read: bool) -> usize;
+	/// Disconnect the socket pointed to by this SocketDescriptor. Once this function returns, no
+	/// more calls to write_event, read_event or disconnect_event may be made with this descriptor.
+	/// No disconnect_event should be generated as a result of this call, though obviously races
+	/// may occur whereby disconnect_socket is called after a call to disconnect_event but prior to
+	/// that event completing.
+	fn disconnect_socket(&mut self);
 }
 
 /// Error for PeerManager errors. If you get one of these, you must disconnect the socket and
@@ -296,7 +302,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 														encode_and_send_msg!(msg, 131);
 														continue;
 													},
-													msgs::ErrorAction::DisconnectPeer => {
+													msgs::ErrorAction::DisconnectPeer { msg: _ } => {
 														return Err(PeerHandleError{ no_connection_possible: false });
 													},
 													msgs::ErrorAction::IgnoreError => {
@@ -723,6 +729,24 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 						}
 						continue;
 					},
+					Event::HandleOutboundMsgGenerationError { ref node_id, ref msg } => {
+						let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
+						});
+						peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
+						Self::do_attempt_write_data(&mut descriptor, peer);
+					},
+					Event::DisconnectPeer { ref node_id, ref msg } => {
+						if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) {
+							if let Some(mut peer) = peers.peers.remove(&descriptor) {
+								if let Some(ref msg) = *msg {
+									peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
+									Self::do_attempt_write_data(&mut descriptor, &mut peer);
+								}
+							}
+							descriptor.disconnect_socket();
+						}
+						self.message_handler.chan_handler.peer_disconnected(&node_id, false);
+					},
 				}
 
 				upstream_events.push(event);
@@ -769,3 +793,83 @@ impl<Descriptor: SocketDescriptor> EventsProvider for PeerManager<Descriptor> {
 		ret
 	}
 }
+
+#[cfg(test)]
+mod tests {
+	use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
+	use util::events;
+	use util::test_utils;
+
+	use secp256k1::Secp256k1;
+	use secp256k1::key::{SecretKey, PublicKey};
+
+	use rand::{thread_rng, Rng};
+
+	use std::sync::{Arc};
+
+	#[derive(PartialEq, Eq, Clone, Hash)]
+	struct FileDescriptor {
+		fd: u16,
+	}
+
+	impl SocketDescriptor for FileDescriptor {
+		fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, _resume_read: bool) -> usize {
+			assert!(write_offset < data.len());
+			data.len() - write_offset
+		}
+
+		fn disconnect_socket(&mut self) {}
+	}
+
+	fn create_network(peer_count: usize) -> Vec<PeerManager<FileDescriptor>> {
+		let secp_ctx = Secp256k1::new();
+		let mut peers = Vec::new();
+		let mut rng = thread_rng();
+
+		for _ in 0..peer_count {
+			let chan_handler = test_utils::TestChannelMessageHandler::new();
+			let router = test_utils::TestRoutingMessageHandler::new();
+			let node_id = {
+				let mut key_slice = [0;32];
+				rng.fill_bytes(&mut key_slice);
+				SecretKey::from_slice(&secp_ctx, &key_slice).unwrap()
+			};
+			let msg_handler = MessageHandler { chan_handler: Arc::new(chan_handler), route_handler: Arc::new(router) };
+			let peer = PeerManager::new(msg_handler, node_id);
+			peers.push(peer);
+		}
+
+		peers
+	}
+
+	fn establish_connection(peer_a: &PeerManager<FileDescriptor>, peer_b: &PeerManager<FileDescriptor>) {
+		let secp_ctx = Secp256k1::new();
+		let their_id = PublicKey::from_secret_key(&secp_ctx, &peer_b.our_node_secret).unwrap();
+		let fd = FileDescriptor { fd: 1};
+		peer_a.new_inbound_connection(fd.clone()).unwrap();
+		peer_a.peers.lock().unwrap().node_id_to_descriptor.insert(their_id, fd.clone());
+	}
+
+	#[test]
+	fn test_disconnect_peer() {
+		// Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and
+		// push an DisconnectPeer event to remove the node flagged by id
+		let mut peers = create_network(2);
+		establish_connection(&peers[0], &peers[1]);
+		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
+
+		let secp_ctx = Secp256k1::new();
+		let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret).unwrap();
+
+		let chan_handler = test_utils::TestChannelMessageHandler::new();
+		chan_handler.pending_events.lock().unwrap().push(events::Event::DisconnectPeer {
+			node_id: their_id,
+			msg: None,
+		});
+		assert_eq!(chan_handler.pending_events.lock().unwrap().len(), 1);
+		peers[0].message_handler.chan_handler = Arc::new(chan_handler);
+
+		peers[0].process_events();
+		assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
+	}
+}

src/util/events.rs

@@ -99,6 +99,16 @@ pub enum Event {
 	BroadcastChannelUpdate {
 		msg: msgs::ChannelUpdate,
 	},
+	/// Used to tell a peer that something is incorrect
+	HandleOutboundMsgGenerationError {
+		node_id: PublicKey,
+		msg: msgs::ErrorMessage,
+	},
+	// Events indicating the network loop should change the state of connection with peer
+	DisconnectPeer  {
+		node_id: PublicKey,
+		msg: Option<msgs::ErrorMessage>,
+	}
 }
 
 pub trait EventsProvider {