X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=e01d2c9b89c4494d68b8720062de63147538a1fc;hb=4fbc0b37688411a7270fce20b44fb43944669823;hp=9715ee965c3374c63494e584721138341452e589;hpb=9606f94caf451d9eb3a690b559e2955486a3e76a;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index 9715ee96..e01d2c9b 100644 --- a/src/ln/peer_handler.rs +++ b/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, + their_global_features: Option, + their_local_features: Option, pending_outbound_buffer: LinkedList>, pending_outbound_buffer_first_msg_offset: usize, @@ -112,7 +117,10 @@ impl PeerManager { 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 PeerManager { 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, @@ -219,7 +230,6 @@ impl PeerManager { } fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: Vec) -> Result { - let mut upstream_events = Vec::new(); let pause_read = { let mut peers = self.peers.lock().unwrap(); let (should_insert_node_id, pause_read) = match peers.peers.get_mut(peer_descriptor) { @@ -280,6 +290,7 @@ impl PeerManager { let act_three = try_potential_handleerror!(peer.channel_encryptor.process_act_two(&peer.pending_read_buffer[..], &self.our_node_secret)).to_vec(); peer.pending_outbound_buffer.push_back(act_three); peer.pending_read_buffer = [0; 18].to_vec(); // Message length header is 18 bytes + peer.pending_read_is_header = true; insert_node_id = Some(peer.their_node_id.unwrap()); encode_and_send_msg!(msgs::Init { @@ -308,6 +319,10 @@ impl PeerManager { assert!(msg_data.len() >= 2); 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{}); + } match msg_type { // Connection control: 16 => { @@ -318,7 +333,15 @@ impl PeerManager { if msg.local_features.requires_unknown_bits() { return Err(PeerHandleError{}); } - //TODO: Store features! + 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); + } }, 17 => { // Error msg @@ -357,11 +380,20 @@ impl PeerManager { 38 => { let msg = try_potential_decodeerror!(msgs::Shutdown::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg)); + let resp_options = try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg)); + if let Some(resp) = resp_options.0 { + encode_and_send_msg!(resp, 38); + } + if let Some(resp) = resp_options.1 { + encode_and_send_msg!(resp, 39); + } }, 39 => { let msg = try_potential_decodeerror!(msgs::ClosingSigned::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg)); + let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg)); + if let Some(resp) = resp_option { + encode_and_send_msg!(resp, 39); + } }, 128 => { @@ -454,11 +486,25 @@ impl PeerManager { None => {} }; + pause_read + }; + + self.process_events(); + + Ok(pause_read) + } + + /// Checks for any events generated by our handlers and processes them. May be needed after eg + /// calls to ChannelManager::process_pending_htlc_forward. + pub fn process_events(&self) { + let mut upstream_events = Vec::new(); + { // TODO: There are some DoS attacks here where you can flood someone's outbound send // buffer by doing things like announcing channels on another node. We should be willing to // drop optional-ish messages when send buffers get full! let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_events(); + let mut peers = self.peers.lock().unwrap(); for event in events_generated.drain(..) { macro_rules! get_peer_for_forwarding { ($node_id: expr, $handle_no_such_peer: block) => { @@ -564,16 +610,12 @@ impl PeerManager { upstream_events.push(event); } - - pause_read - }; + } let mut pending_events = self.pending_events.lock().unwrap(); for event in upstream_events.drain(..) { pending_events.push(event); } - - Ok(pause_read) } /// Indicates that the given socket descriptor's connection is now closed.