Clean up peer_handler error handling a bit, clean up channel error handling a ton

fuzz/travis-fuzz.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
-cargo install honggfuzz
-set +e
+set -e
+cargo install --force honggfuzz
 for TARGET in fuzz_targets/*; do
     FILENAME=$(basename $TARGET)
 	FILE="${FILENAME%.*}"

src/ln/chan_utils.rs

@@ -230,6 +230,6 @@ pub fn get_htlc_redeemscript_with_explicit_keys(htlc: &HTLCOutputInCommitment, a
 /// note here that 'a_revocation_key' is generated using b_revocation_basepoint and a's
 /// commitment secret. 'htlc' does *not* need to have its previous_output_index filled.
 #[inline]
-pub fn get_htlc_redeemscript(htlc: &HTLCOutputInCommitment, keys: &TxCreationKeys, offered: bool) -> Script {
-	get_htlc_redeemscript_with_explicit_keys(htlc, &keys.a_htlc_key, &keys.b_htlc_key, &keys.revocation_key, offered)
+pub fn get_htlc_redeemscript(htlc: &HTLCOutputInCommitment, keys: &TxCreationKeys) -> Script {
+	get_htlc_redeemscript_with_explicit_keys(htlc, &keys.a_htlc_key, &keys.b_htlc_key, &keys.revocation_key, htlc.offered)
 }

src/ln/channelmanager.rs

@@ -1121,7 +1121,7 @@ impl ChannelMessageHandler for ChannelManager {
 			($msg: expr, $err_code: expr, $data: expr) => {
 				return Err(msgs::HandleError {
 					err: $msg,
-					msg: Some(msgs::ErrorMessage::UpdateFailHTLC {
+					msg: Some(msgs::ErrorAction::UpdateFailHTLC {
 						msg: msgs::UpdateFailHTLC {
 							channel_id: msg.channel_id,
 							htlc_id: msg.htlc_id,
@@ -1481,6 +1481,37 @@ impl ChannelMessageHandler for ChannelManager {
 		pending_events.push(events::Event::BroadcastChannelAnnouncement { msg: chan_announcement, update_msg: chan_update });
 		Ok(())
 	}
+
+	fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool) {
+		let mut channel_state_lock = self.channel_state.lock().unwrap();
+		let channel_state = channel_state_lock.borrow_parts();
+		let short_to_id = channel_state.short_to_id;
+		if no_connection_possible {
+			channel_state.by_id.retain(move |_, chan| {
+				if chan.get_their_node_id() == *their_node_id {
+					match chan.get_short_channel_id() {
+						Some(short_id) => {
+							short_to_id.remove(&short_id);
+						},
+						None => {},
+					}
+					//TODO: get the latest commitment tx, any HTLC txn built on top of it, etc out
+					//of the channel and throw those into the announcement blackhole.
+					false
+				} else {
+					true
+				}
+			});
+		} else {
+			for chan in channel_state.by_id {
+				if chan.1.get_their_node_id() == *their_node_id {
+					//TODO: mark channel disabled (and maybe announce such after a timeout). Also
+					//fail and wipe any uncommitted outbound HTLCs as those are considered after
+					//reconnect.
+				}
+			}
+		}
+	}
 }
 
 #[cfg(test)]

src/ln/msgs.rs

@@ -342,16 +342,19 @@ pub struct ChannelUpdate {
 }
 
 /// Used to put an error message in a HandleError
-pub enum ErrorMessage {
+pub enum ErrorAction {
+	/// Indicates an inbound HTLC add resulted in a failure, and the UpdateFailHTLC provided in msg
+	/// should be sent back to the sender.
 	UpdateFailHTLC {
 		msg: UpdateFailHTLC
 	},
+	/// The peer took some action which made us think they were useless. Disconnect them.
 	DisconnectPeer {},
 }
 
 pub struct HandleError { //TODO: rename me
 	pub err: &'static str,
-	pub msg: Option<ErrorMessage>, //TODO: Move into an Action enum and require it!
+	pub msg: Option<ErrorAction>, //TODO: Make this required and rename it
 }
 
 /// A trait to describe an object which can receive channel messages. Messages MAY be called in
@@ -381,6 +384,13 @@ pub trait ChannelMessageHandler : events::EventsProvider {
 
 	// Channel-to-announce:
 	fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &AnnouncementSignatures) -> Result<(), HandleError>;
+
+	// Informational:
+	/// Indicates a connection to the peer failed/an existing connection was lost. If no connection
+	/// is believed to be possible in the future (eg they're sending us messages we don't
+	/// understand or indicate they require unknown feature bits), no_connection_possible is set
+	/// and any outstanding channels should be failed.
+	fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool);
 }
 
 pub trait RoutingMessageHandler {

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", msg: Some(msgs::ErrorMessage::DisconnectPeer{})});
+			return Err(HandleError{err: "Bad MAC", msg: Some(msgs::ErrorAction::DisconnectPeer{})});
 		}
 		Ok(())
 	}
@@ -195,11 +195,11 @@ impl PeerChannelEncryptor {
 		assert_eq!(act.len(), 50);
 
 		if act[0] != 0 {
-			return Err(HandleError{err: "Unknown handshake version number", msg: Some(msgs::ErrorMessage::DisconnectPeer{})});
+			return Err(HandleError{err: "Unknown handshake version number", msg: Some(msgs::ErrorAction::DisconnectPeer{})});
 		}
 
 		let their_pub = match PublicKey::from_slice(secp_ctx, &act[1..34]) {
-			Err(_) => return Err(HandleError{err: "Invalid public key", msg: Some(msgs::ErrorMessage::DisconnectPeer{})}),
+			Err(_) => return Err(HandleError{err: "Invalid public key", msg: Some(msgs::ErrorAction::DisconnectPeer{})}),
 			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", msg: Some(msgs::ErrorMessage::DisconnectPeer{})});
+							return Err(HandleError{err: "Unknown handshake version number", msg: Some(msgs::ErrorAction::DisconnectPeer{})});
 						}
 
 						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", msg: Some(msgs::ErrorMessage::DisconnectPeer{})}),
+							Err(_) => return Err(HandleError{err: "Bad node_id from peer", msg: Some(msgs::ErrorAction::DisconnectPeer{})}),
 						});
 
 						let mut sha = Sha256::new();

src/ln/peer_handler.rs

@@ -40,16 +40,21 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone {
 /// generate no further read/write_events for the descriptor, only triggering a single
 /// disconnect_event (unless it was provided in response to a new_*_connection event, in which case
 /// no such disconnect_event must be generated and the socket be silently disconencted).
-pub struct PeerHandleError {}
+pub struct PeerHandleError {
+	no_connection_possible: bool,
+}
 impl fmt::Debug for PeerHandleError {
 	fn fmt(&self, formatter: &mut fmt::Formatter) -> Result<(), fmt::Error> {
-		formatter.write_str("Peer Send Invalid Data")
+		formatter.write_str("Peer Sent Invalid Data")
 	}
 }
 
 struct Peer {
 	channel_encryptor: PeerChannelEncryptor,
+	outbound: bool,
 	their_node_id: Option<PublicKey>,
+	their_global_features: Option<msgs::GlobalFeatures>,
+	their_local_features: Option<msgs::LocalFeatures>,
 
 	pending_outbound_buffer: LinkedList<Vec<u8>>,
 	pending_outbound_buffer_first_msg_offset: usize,
@@ -112,7 +117,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 		let mut peers = self.peers.lock().unwrap();
 		if peers.peers.insert(descriptor, Peer {
 			channel_encryptor: peer_encryptor,
+			outbound: true,
 			their_node_id: Some(their_node_id),
+			their_global_features: None,
+			their_local_features: None,
 
 			pending_outbound_buffer: LinkedList::new(),
 			pending_outbound_buffer_first_msg_offset: 0,
@@ -141,7 +149,10 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 		let mut peers = self.peers.lock().unwrap();
 		if peers.peers.insert(descriptor, Peer {
 			channel_encryptor: peer_encryptor,
+			outbound: false,
 			their_node_id: None,
+			their_global_features: None,
+			their_local_features: None,
 
 			pending_outbound_buffer: LinkedList::new(),
 			pending_outbound_buffer_first_msg_offset: 0,
@@ -212,7 +223,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 		match self.do_read_event(peer_descriptor, data) {
 			Ok(res) => Ok(res),
 			Err(e) => {
-				self.disconnect_event(peer_descriptor);
+				self.disconnect_event_internal(peer_descriptor, e.no_connection_possible);
 				Err(e)
 			}
 		}
@@ -227,38 +238,7 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 					assert!(peer.pending_read_buffer.len() > 0);
 					assert!(peer.pending_read_buffer.len() > peer.pending_read_buffer_pos);
 
-					macro_rules! try_potential_handleerror {
-						($thing: expr) => {
-							match $thing {
-								Ok(x) => x,
-								Err(_e) => {
-									//TODO: Handle e appropriately!
-									return Err(PeerHandleError{});
-								}
-							};
-						}
-					}
-
-					macro_rules! try_potential_decodeerror {
-						($thing: expr) => {
-							match $thing {
-								Ok(x) => x,
-								Err(_e) => {
-									//TODO: Handle e?
-									return Err(PeerHandleError{});
-								}
-							};
-						}
-					}
-
-					macro_rules! encode_and_send_msg {
-						($msg: expr, $msg_code: expr) => {
-							peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
-						}
-					}
-
 					let mut insert_node_id = None;
-
 					let mut read_pos = 0;
 					while read_pos < data.len() {
 						{
@@ -267,7 +247,52 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 							read_pos += data_to_copy;
 							peer.pending_read_buffer_pos += data_to_copy;
 						}
+
 						if peer.pending_read_buffer_pos == peer.pending_read_buffer.len() {
+							peer.pending_read_buffer_pos = 0;
+
+							macro_rules! encode_and_send_msg {
+								($msg: expr, $msg_code: expr) => {
+									peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!($msg, $msg_code)[..]));
+								}
+							}
+
+							macro_rules! try_potential_handleerror {
+								($thing: expr) => {
+									match $thing {
+										Ok(x) => x,
+										Err(e) => {
+											// TODO: Log e.err
+											if let Some(action) = e.msg {
+												match action {
+													msgs::ErrorAction::UpdateFailHTLC { msg } => {
+														encode_and_send_msg!(msg, 131);
+														continue;
+													},
+													msgs::ErrorAction::DisconnectPeer {} => {
+														return Err(PeerHandleError{ no_connection_possible: false });
+													},
+												}
+											} else {
+												return Err(PeerHandleError{ no_connection_possible: false });
+											}
+										}
+									};
+								}
+							}
+
+							macro_rules! try_potential_decodeerror {
+								($thing: expr) => {
+									match $thing {
+										Ok(x) => x,
+										Err(_e) => {
+											//TODO: Handle e?
+											return Err(PeerHandleError{ no_connection_possible: false });
+										}
+									};
+								}
+							}
+
 							let next_step = peer.channel_encryptor.get_noise_step();
 							match next_step {
 								NextNoiseStep::ActOne => {
@@ -300,27 +325,41 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 										peer.pending_read_buffer = Vec::with_capacity(msg_len as usize + 16);
 										peer.pending_read_buffer.resize(msg_len as usize + 16, 0);
 										if msg_len < 2 { // Need at least the message type tag
-											return Err(PeerHandleError{});
+											return Err(PeerHandleError{ no_connection_possible: false });
 										}
 										peer.pending_read_is_header = false;
 									} else {
 										let msg_data = try_potential_handleerror!(peer.channel_encryptor.decrypt_message(&peer.pending_read_buffer[..]));
 										assert!(msg_data.len() >= 2);
 
+										// Reset read buffer
+										peer.pending_read_buffer = [0; 18].to_vec();
+										peer.pending_read_is_header = true;
+
 										let msg_type = byte_utils::slice_to_be16(&msg_data[0..2]);
+										if msg_type != 16 && peer.their_global_features.is_none() {
+											// Need an init message as first message
+											return Err(PeerHandleError{ no_connection_possible: false });
+										}
 										match msg_type {
 											// Connection control:
 											16 => {
 												let msg = try_potential_decodeerror!(msgs::Init::decode(&msg_data[2..]));
 												if msg.global_features.requires_unknown_bits() {
-													return Err(PeerHandleError{});
+													return Err(PeerHandleError{ no_connection_possible: true });
 												}
 												if msg.local_features.requires_unknown_bits() {
-													return Err(PeerHandleError{});
+													return Err(PeerHandleError{ no_connection_possible: true });
+												}
+												peer.their_global_features = Some(msg.global_features);
+												peer.their_local_features = Some(msg.local_features);
+
+												if !peer.outbound {
+													encode_and_send_msg!(msgs::Init {
+														global_features: msgs::GlobalFeatures::new(),
+														local_features: msgs::LocalFeatures::new(),
+													}, 16);
 												}
-												//TODO: Store features (and check that we've
-												//received Init prior to any other messages)!
-												//TODO: Respond to Init with Init if we're inbound.
 											},
 											17 => {
 												// Error msg
@@ -439,18 +478,13 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 											},
 											_ => {
 												if (msg_type & 1) == 0 {
-													//TODO: Fail all channels. Kill the peer!
-													return Err(PeerHandleError{});
+													return Err(PeerHandleError{ no_connection_possible: true });
 												}
 											},
 										}
-
-										peer.pending_read_buffer = [0; 18].to_vec();
-										peer.pending_read_is_header = true;
 									}
 								}
 							}
-							peer.pending_read_buffer_pos = 0;
 						}
 					}
 
@@ -602,18 +636,22 @@ impl<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
 	/// but must NOT be called if a PeerHandleError was provided out of a new_*_connection event!
 	/// Panics if the descriptor was not previously registered in a successful new_*_connection event.
 	pub fn disconnect_event(&self, descriptor: &Descriptor) {
+		self.disconnect_event_internal(descriptor, false);
+	}
+
+	fn disconnect_event_internal(&self, descriptor: &Descriptor, no_connection_possible: bool) {
 		let mut peers = self.peers.lock().unwrap();
 		let peer_option = peers.peers.remove(descriptor);
 		match peer_option {
 			None => panic!("Descriptor for disconnect_event is not already known to PeerManager"),
 			Some(peer) => {
 				match peer.their_node_id {
-					Some(node_id) => { peers.node_id_to_descriptor.remove(&node_id); },
+					Some(node_id) => {
+						peers.node_id_to_descriptor.remove(&node_id);
+						self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible);
+					},
 					None => {}
 				}
-				//TODO: Notify the chan_handler that this node disconnected, and do something about
-				//handling response messages that were queued for sending (maybe the send buffer
-				//needs to be unencrypted?)
 			}
 		};
 	}